



"""simple docstring"""


from sklearn.metrics import fa_score, matthews_corrcoef

import datasets

from .record_evaluation import evaluate as evaluate_record


__A   :							Dict										=   "\\n@article{wang2019superglue,\n  title={SuperGLUE: A Stickier Benchmark for General-Purpose Language Understanding Systems},\n  author={Wang, Alex and Pruksachatkun, Yada and Nangia, Nikita and Singh, Amanpreet and Michael, Julian and Hill, Felix and Levy, Omer and Bowman, Samuel R},\n  journal={arXiv preprint arXiv:1905.00537},\n  year={2019}\n}\n"

__A   :							Union[str, Any]										=   "\\nSuperGLUE (https://super.gluebenchmark.com/) is a new benchmark styled after\nGLUE with a new set of more difficult language understanding tasks, improved\nresources, and a new public leaderboard.\n"

__A   :							List[str]										=   "\nCompute SuperGLUE evaluation metric associated to each SuperGLUE dataset.\nArgs:\n    predictions: list of predictions to score. Depending on the SuperGlUE subset:\n        - for \'record\': list of question-answer dictionaries with the following keys:\n            - \'idx\': index of the question as specified by the dataset\n            - \'prediction_text\': the predicted answer text\n        - for \'multirc\': list of question-answer dictionaries with the following keys:\n            - \'idx\': index of the question-answer pair as specified by the dataset\n            - \'prediction\': the predicted answer label\n        - otherwise: list of predicted labels\n    references: list of reference labels. Depending on the SuperGLUE subset:\n        - for \'record\': list of question-answers dictionaries with the following keys:\n            - \'idx\': index of the question as specified by the dataset\n            - \'answers\': list of possible answers\n        - otherwise: list of reference labels\nReturns: depending on the SuperGLUE subset:\n    - for \'record\':\n        - \'exact_match\': Exact match between answer and gold answer\n        - \'f1\': F1 score\n    - for \'multirc\':\n        - \'exact_match\': Exact match between answer and gold answer\n        - \'f1_m\': Per-question macro-F1 score\n        - \'f1_a\': Average F1 score over all answers\n    - for \'axb\':\n        \'matthews_correlation\': Matthew Correlation\n    - for \'cb\':\n        - \'accuracy\': Accuracy\n        - \'f1\': F1 score\n    - for all others:\n        - \'accuracy\': Accuracy\nExamples:\n\n    >>> super_glue_metric = datasets.load_metric(\'super_glue\', \'copa\')  # any of [\"copa\", \"rte\", \"wic\", \"wsc\", \"wsc.fixed\", \"boolq\", \"axg\"]\n    >>> predictions = [0, 1]\n    >>> references = [0, 1]\n    >>> results = super_glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {\'accuracy\': 1.0}\n\n    >>> super_glue_metric = datasets.load_metric(\'super_glue\', \'cb\')\n    >>> predictions = [0, 1]\n    >>> references = [0, 1]\n    >>> results = super_glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {\'accuracy\': 1.0, \'f1\': 1.0}\n\n    >>> super_glue_metric = datasets.load_metric(\'super_glue\', \'record\')\n    >>> predictions = [{\'idx\': {\'passage\': 0, \'query\': 0}, \'prediction_text\': \'answer\'}]\n    >>> references = [{\'idx\': {\'passage\': 0, \'query\': 0}, \'answers\': [\'answer\', \'another_answer\']}]\n    >>> results = super_glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {\'exact_match\': 1.0, \'f1\': 1.0}\n\n    >>> super_glue_metric = datasets.load_metric(\'super_glue\', \'multirc\')\n    >>> predictions = [{\'idx\': {\'answer\': 0, \'paragraph\': 0, \'question\': 0}, \'prediction\': 0}, {\'idx\': {\'answer\': 1, \'paragraph\': 2, \'question\': 3}, \'prediction\': 1}]\n    >>> references = [0, 1]\n    >>> results = super_glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {\'exact_match\': 1.0, \'f1_m\': 1.0, \'f1_a\': 1.0}\n\n    >>> super_glue_metric = datasets.load_metric(\'super_glue\', \'axb\')\n    >>> references = [0, 1]\n    >>> predictions = [0, 1]\n    >>> results = super_glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {\'matthews_correlation\': 1.0}\n"







def  lowercase	(      _SCREAMING_SNAKE_CASE   :						List[Any]					,     _SCREAMING_SNAKE_CASE   :						int				):






				'''simple docstring'''



				return float((preds == labels).mean()				)







def  lowercase	(      _SCREAMING_SNAKE_CASE   :						List[Any]					,     _SCREAMING_SNAKE_CASE   :						Tuple					,     _SCREAMING_SNAKE_CASE   :						Optional[Any]="binary"				):






				'''simple docstring'''



				_UpperCAmelCase					=    simple_accuracy(_SCREAMING_SNAKE_CASE					,     _SCREAMING_SNAKE_CASE				)
				_UpperCAmelCase					=    float(fa_score(y_true=_SCREAMING_SNAKE_CASE					,     y_pred=_SCREAMING_SNAKE_CASE					,     average=_SCREAMING_SNAKE_CASE				)				)
				return {
				    "accuracy": acc,
				    "f1": fa,
				}







def  lowercase	(      _SCREAMING_SNAKE_CASE   :						Any					,     _SCREAMING_SNAKE_CASE   :						Union[str, Any]				):






				'''simple docstring'''



				_UpperCAmelCase					=    {}
				for id_pred, label in zip(_SCREAMING_SNAKE_CASE					,     _SCREAMING_SNAKE_CASE				):
								_UpperCAmelCase					=    f'{id_pred["idx"]["paragraph"]}-{id_pred["idx"]["question"]}'
								_UpperCAmelCase					=    id_pred['''prediction''']
								if question_id in question_map:
												question_map[question_id].append((pred, label)				)
								else:
												_UpperCAmelCase					=    [(pred, label)]
				_UpperCAmelCase		,						_UpperCAmelCase					=    [], []
				for question, preds_labels in question_map.items():
								_UpperCAmelCase		,						_UpperCAmelCase					=    zip(*_SCREAMING_SNAKE_CASE				)
								_UpperCAmelCase					=    fa_score(y_true=_SCREAMING_SNAKE_CASE					,     y_pred=_SCREAMING_SNAKE_CASE					,     average='''macro'''				)
								fas.append(_SCREAMING_SNAKE_CASE				)
								_UpperCAmelCase					=    int(sum(pred == label for pred, label in preds_labels				) == len(_SCREAMING_SNAKE_CASE				)				)
								ems.append(_SCREAMING_SNAKE_CASE				)
				_UpperCAmelCase					=    float(sum(_SCREAMING_SNAKE_CASE				) / len(_SCREAMING_SNAKE_CASE				)				)
				_UpperCAmelCase					=    sum(_SCREAMING_SNAKE_CASE				) / len(_SCREAMING_SNAKE_CASE				)
				_UpperCAmelCase					=    float(fa_score(y_true=_SCREAMING_SNAKE_CASE					,     y_pred=[id_pred['''prediction'''] for id_pred in ids_preds]				)				)
				return {"exact_match": em, "f1_m": fa_m, "f1_a": fa_a}
@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION			,							_KWARGS_DESCRIPTION)
class 			_a (					datasets.Metric):
			"""simple docstring"""



			def  lowercase__						(		self     :    List[str]   )->int:
							if self.config_name not in [
							    "boolq",
							    "cb",
							    "copa",
							    "multirc",
							    "record",
							    "rte",
							    "wic",
							    "wsc",
							    "wsc.fixed",
							    "axb",
							    "axg",
							]:
											raise KeyError(
											    '''You should supply a configuration name selected in '''
											    '''["boolq", "cb", "copa", "multirc", "record", "rte", "wic", "wsc", "wsc.fixed", "axb", "axg",]'''   )
							return datasets.MetricInfo(
							    description=_DESCRIPTION				,   citation=_CITATION				,   inputs_description=_KWARGS_DESCRIPTION				,   features=datasets.Features(self._get_feature_types()   )				,   codebase_urls=[]				,   reference_urls=[]				,   format='''numpy''' if not self.config_name == '''record''' and not self.config_name == '''multirc''' else None				,   )



			def  lowercase__						(		self     :    List[Any]   )->Optional[Any]:
							if self.config_name == "record":
											return {
											    "predictions": {
											        "idx": {
											            "passage": datasets.Value('''int64'''   ),
											            "query": datasets.Value('''int64'''   ),
											        },
											        "prediction_text": datasets.Value('''string'''   ),
											    },
											    "references": {
											        "idx": {
											            "passage": datasets.Value('''int64'''   ),
											            "query": datasets.Value('''int64'''   ),
											        },
											        "answers": datasets.Sequence(datasets.Value('''string'''   )   ),
											    },
											}
							elif self.config_name == "multirc":
											return {
											    "predictions": {
											        "idx": {
											            "answer": datasets.Value('''int64'''   ),
											            "paragraph": datasets.Value('''int64'''   ),
											            "question": datasets.Value('''int64'''   ),
											        },
											        "prediction": datasets.Value('''int64'''   ),
											    },
											    "references": datasets.Value('''int64'''   ),
											}
							else:
											return {
											    "predictions": datasets.Value('''int64'''   ),
											    "references": datasets.Value('''int64'''   ),
											}



			def  lowercase__						(		self     :    Any				,   __UpperCamelCase     :    int				,   __UpperCamelCase     :    Optional[Any]   )->Tuple:
							if self.config_name == "axb":
											return {"matthews_correlation": matthews_corrcoef(lowerCamelCase_				,   lowerCamelCase_   )}
							elif self.config_name == "cb":
											return acc_and_fa(lowerCamelCase_				,   lowerCamelCase_				,   fa_avg='''macro'''   )
							elif self.config_name == "record":
											_UpperCAmelCase					=    [
											    {
											        '''qas''': [
											            {'''id''': ref['''idx''']['''query'''], '''answers''': [{'''text''': ans} for ans in ref['''answers''']]}
											            for ref in references
											        ]
											    }
											]
											_UpperCAmelCase					=    {pred['''idx''']['''query''']: pred['''prediction_text'''] for pred in predictions}
											return evaluate_record(lowerCamelCase_				,   lowerCamelCase_   )[0]
							elif self.config_name == "multirc":
											return evaluate_multirc(lowerCamelCase_				,   lowerCamelCase_   )
							elif self.config_name in ["copa", "rte", "wic", "wsc", "wsc.fixed", "boolq", "axg"]:
											return {"accuracy": simple_accuracy(lowerCamelCase_				,   lowerCamelCase_   )}
							else:
											raise KeyError(
											    '''You should supply a configuration name selected in '''
											    '''["boolq", "cb", "copa", "multirc", "record", "rte", "wic", "wsc", "wsc.fixed", "axb", "axg",]'''   )








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


__A   :							List[str]										=   logging.get_logger(__name__)
@add_end_docstrings(lowerCAmelCase)
class 			_a (					lowerCAmelCase):
			"""simple docstring"""



			def __init__(		self     :    Union[str, Any]				,   **__UpperCamelCase     :    Optional[Any]   )->List[Any]:
							super().__init__(**__UpperCamelCase   )

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



			def __call__(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[np.ndarray, bytes, str]				,   **__UpperCamelCase     :    Tuple   )->List[str]:
							return super().__call__(__UpperCamelCase				,   **__UpperCamelCase   )



			def  lowercase__						(		self     :    Any				,   **__UpperCamelCase     :    Any   )->Union[str, Any]:
							_UpperCAmelCase					=    {}
							if "candidate_labels" in kwargs:
											_UpperCAmelCase					=    kwargs['''candidate_labels''']
							if "hypothesis_template" in kwargs:
											_UpperCAmelCase					=    kwargs['''hypothesis_template''']

							return preprocess_params, {}, {}



			def  lowercase__						(		self     :    Optional[int]				,   __UpperCamelCase     :    Union[str, Any]				,   __UpperCamelCase     :    List[Any]=None				,   __UpperCamelCase     :    Tuple="This is a sound of {}."   )->int:
							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											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
															_UpperCAmelCase					=    requests.get(__UpperCamelCase   ).content
											else:
															with open(__UpperCamelCase				,   '''rb'''   ) as f:
																			_UpperCAmelCase					=    f.read()

							if isinstance(__UpperCamelCase				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    ffmpeg_read(__UpperCamelCase				,   self.feature_extractor.sampling_rate   )

							if not isinstance(__UpperCamelCase				,   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'''   )

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



			def  lowercase__						(		self     :    Union[str, Any]				,   __UpperCamelCase     :    Optional[int]   )->Any:
							_UpperCAmelCase					=    model_inputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_inputs.pop('''text_inputs'''   )
							if isinstance(text_inputs[0]				,   __UpperCamelCase   ):
											_UpperCAmelCase					=    text_inputs[0]
							else:
											# Batching case.
											_UpperCAmelCase					=    text_inputs[0][0]

							_UpperCAmelCase					=    self.model(**__UpperCamelCase				,   **__UpperCamelCase   )

							_UpperCAmelCase					=    {
							    '''candidate_labels''': candidate_labels,
							    '''logits''': outputs.logits_per_audio,
							}
							return model_outputs



			def  lowercase__						(		self     :    List[str]				,   __UpperCamelCase     :    Any   )->List[Any]:
							_UpperCAmelCase					=    model_outputs.pop('''candidate_labels'''   )
							_UpperCAmelCase					=    model_outputs['''logits'''][0]

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

							_UpperCAmelCase					=    [
							    {'''score''': score, '''label''': candidate_label}
							    for score, candidate_label in sorted(zip(__UpperCamelCase				,   __UpperCamelCase   )				,   key=lambda __UpperCamelCase   : -x[0]   )
							]
							return result



