


import json
import os
from typing import Optional

import numpy as np

from ...feature_extraction_utils import BatchFeature
from ...processing_utils import ProcessorMixin
from ...utils import logging
from ...utils.hub import get_file_from_repo
from ..auto import AutoTokenizer


__a     =			logging.get_logger(__name__)






class   __lowercase				(  __snake_case  ):
						UpperCamelCase								=       '''AutoTokenizer'''
						UpperCamelCase								=       ['''tokenizer''']

						UpperCamelCase								=       {
						    '''semantic_prompt''': 1,
						    '''coarse_prompt''': 2,
						    '''fine_prompt''': 2,
						}





						def __init__(		self						:					Optional[int]					,       __lowerCamelCase						:					Optional[int]					,       __lowerCamelCase						:					Any=None       )    ->   Any:






									"""simple docstring"""







									super().__init__(__lowerCamelCase       )

									UpperCAmelCase							=     speaker_embeddings





						@classmethod
						def        _lowercase				(		cls						:					Tuple					,       __lowerCamelCase						:					List[str]					,       __lowerCamelCase						:					str="speaker_embeddings_path.json"					,       **__lowerCamelCase						:					Any       )    ->   str:






									"""simple docstring"""







									if speaker_embeddings_dict_path is not None:
												UpperCAmelCase							=     get_file_from_repo(
												    __lowerCamelCase					,       __lowerCamelCase					,       subfolder=kwargs.pop("""subfolder"""					,       __lowerCamelCase       )					,       cache_dir=kwargs.pop("""cache_dir"""					,       __lowerCamelCase       )					,       force_download=kwargs.pop("""force_download"""					,       __lowerCamelCase       )					,       proxies=kwargs.pop("""proxies"""					,       __lowerCamelCase       )					,       resume_download=kwargs.pop("""resume_download"""					,       __lowerCamelCase       )					,       local_files_only=kwargs.pop("""local_files_only"""					,       __lowerCamelCase       )					,       use_auth_token=kwargs.pop("""use_auth_token"""					,       __lowerCamelCase       )					,       revision=kwargs.pop("""revision"""					,       __lowerCamelCase       )					,       )
												if speaker_embeddings_path is None:
															logger.warning(
															    F"""`{os.path.join(__lowerCamelCase					,       __lowerCamelCase       )}` does not exists
                    , no preloaded speaker embeddings will be used - Make sure to provide a correct path to the json
                    dictionnary if wanted, otherwise set `speaker_embeddings_dict_path=None`."""       )
															UpperCAmelCase							=     None
												else:
															with open(__lowerCamelCase       ) as speaker_embeddings_json:
																		UpperCAmelCase							=     json.load(__lowerCamelCase       )
									else:
												UpperCAmelCase							=     None

									UpperCAmelCase							=     AutoTokenizer.from_pretrained(__lowerCamelCase					,       **__lowerCamelCase       )

									return cls(tokenizer=__lowerCamelCase					,       speaker_embeddings=__lowerCamelCase       )





						def        _lowercase				(		self						:					Optional[int]					,       __lowerCamelCase						:					Optional[Any]					,       __lowerCamelCase						:					int="speaker_embeddings_path.json"					,       __lowerCamelCase						:					Any="speaker_embeddings"					,       __lowerCamelCase						:					bool = False					,       **__lowerCamelCase						:					Any					,       )    ->   str:






									"""simple docstring"""







									if self.speaker_embeddings is not None:
												os.makedirs(os.path.join(__lowerCamelCase					,       __lowerCamelCase					,       """v2"""       )					,       exist_ok=__lowerCamelCase       )

												UpperCAmelCase							=     {}

												UpperCAmelCase							=     save_directory

												for prompt_key in self.speaker_embeddings:
															if prompt_key != "repo_or_path":
																		UpperCAmelCase							=     self._load_voice_preset(__lowerCamelCase       )

																		UpperCAmelCase							=     {}
																		for key in self.speaker_embeddings[prompt_key]:
																					np.save(
																					    os.path.join(
																					        embeddings_dict["""repo_or_path"""]					,       __lowerCamelCase					,       F"""{prompt_key}_{key}"""       )					,       voice_preset[key]					,       allow_pickle=__lowerCamelCase					,       )
																					UpperCAmelCase							=     os.path.join(__lowerCamelCase					,       F"""{prompt_key}_{key}.npy"""       )

																		UpperCAmelCase							=     tmp_dict

												with open(os.path.join(__lowerCamelCase					,       __lowerCamelCase       )					,       """w"""       ) as fp:
															json.dump(__lowerCamelCase					,       __lowerCamelCase       )

									super().save_pretrained(__lowerCamelCase					,       __lowerCamelCase					,       **__lowerCamelCase       )





						def        _lowercase				(		self						:					int					,       __lowerCamelCase						:					str = None					,       **__lowerCamelCase						:					Tuple       )    ->   Union[str, Any]:






									"""simple docstring"""







									UpperCAmelCase							=     self.speaker_embeddings[voice_preset]

									UpperCAmelCase							=     {}
									for key in ["semantic_prompt", "coarse_prompt", "fine_prompt"]:
												if key not in voice_preset_paths:
															raise ValueError(
															    F"""Voice preset unrecognized, missing {key} as a key in self.speaker_embeddings[{voice_preset}]."""       )

												UpperCAmelCase							=     get_file_from_repo(
												    self.speaker_embeddings.get("""repo_or_path"""					,       """/"""       )					,       voice_preset_paths[key]					,       subfolder=kwargs.pop("""subfolder"""					,       __lowerCamelCase       )					,       cache_dir=kwargs.pop("""cache_dir"""					,       __lowerCamelCase       )					,       force_download=kwargs.pop("""force_download"""					,       __lowerCamelCase       )					,       proxies=kwargs.pop("""proxies"""					,       __lowerCamelCase       )					,       resume_download=kwargs.pop("""resume_download"""					,       __lowerCamelCase       )					,       local_files_only=kwargs.pop("""local_files_only"""					,       __lowerCamelCase       )					,       use_auth_token=kwargs.pop("""use_auth_token"""					,       __lowerCamelCase       )					,       revision=kwargs.pop("""revision"""					,       __lowerCamelCase       )					,       )
												if path is None:
															raise ValueError(
															    F"""`{os.path.join(self.speaker_embeddings.get('repo_or_path'					,       '/'       )					,       voice_preset_paths[key]       )}` does not exists
                    , no preloaded voice preset will be used - Make sure to provide correct paths to the {voice_preset}
                    embeddings."""       )

												UpperCAmelCase							=     np.load(__lowerCamelCase       )

									return voice_preset_dict





						def        _lowercase				(		self						:					Any					,       __lowerCamelCase						:					Optional[dict] = None       )    ->   Any:






									"""simple docstring"""







									for key in ["semantic_prompt", "coarse_prompt", "fine_prompt"]:
												if key not in voice_preset:
															raise ValueError(F"""Voice preset unrecognized, missing {key} as a key."""       )

												if not isinstance(voice_preset[key]					,       np.ndarray       ):
															raise ValueError(F"""{key} voice preset must be a {str(self.preset_shape[key]       )}D ndarray."""       )

												if len(voice_preset[key].shape       ) != self.preset_shape[key]:
															raise ValueError(F"""{key} voice preset must be a {str(self.preset_shape[key]       )}D ndarray."""       )





						def __call__(		self						:					int					,       __lowerCamelCase						:					Optional[Any]=None					,       __lowerCamelCase						:					List[str]=None					,       __lowerCamelCase						:					str="pt"					,       __lowerCamelCase						:					Any=2_5_6					,       __lowerCamelCase						:					List[str]=False					,       __lowerCamelCase						:					Union[str, Any]=True					,       __lowerCamelCase						:					Optional[Any]=False					,       **__lowerCamelCase						:					Optional[int]					,       )    ->   List[Any]:






									"""simple docstring"""







									if voice_preset is not None and not isinstance(__lowerCamelCase					,       __lowerCamelCase       ):
												if (
												    isinstance(__lowerCamelCase					,       __lowerCamelCase       )
												    and self.speaker_embeddings is not None
												    and voice_preset in self.speaker_embeddings
												):
															UpperCAmelCase							=     self._load_voice_preset(__lowerCamelCase       )

												else:
															if isinstance(__lowerCamelCase					,       __lowerCamelCase       ) and not voice_preset.endswith(""".npz"""       ):
																		UpperCAmelCase							=     voice_preset + """.npz"""

															UpperCAmelCase							=     np.load(__lowerCamelCase       )

									if voice_preset is not None:
												self._validate_voice_preset_dict(__lowerCamelCase					,       **__lowerCamelCase       )
												UpperCAmelCase							=     BatchFeature(data=__lowerCamelCase					,       tensor_type=__lowerCamelCase       )

									UpperCAmelCase							=     self.tokenizer(
									    __lowerCamelCase					,       return_tensors=__lowerCamelCase					,       padding="""max_length"""					,       max_length=__lowerCamelCase					,       return_attention_mask=__lowerCamelCase					,       return_token_type_ids=__lowerCamelCase					,       add_special_tokens=__lowerCamelCase					,       **__lowerCamelCase					,       )

									if voice_preset is not None:
												UpperCAmelCase							=     voice_preset

									return encoded_text







from sklearn.metrics import fa_score, matthews_corrcoef

import datasets

from .record_evaluation import evaluate as evaluate_record


__a     =			"""\
@article{wang2019superglue,
  title={SuperGLUE: A Stickier Benchmark for General-Purpose Language Understanding Systems},
  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},
  journal={arXiv preprint arXiv:1905.00537},
  year={2019}
}
"""

__a     =			"""\
SuperGLUE (https://super.gluebenchmark.com/) is a new benchmark styled after
GLUE with a new set of more difficult language understanding tasks, improved
resources, and a new public leaderboard.
"""

__a     =			"""
Compute SuperGLUE evaluation metric associated to each SuperGLUE dataset.
Args:
    predictions: list of predictions to score. Depending on the SuperGlUE subset:
        - for 'record': list of question-answer dictionaries with the following keys:
            - 'idx': index of the question as specified by the dataset
            - 'prediction_text': the predicted answer text
        - for 'multirc': list of question-answer dictionaries with the following keys:
            - 'idx': index of the question-answer pair as specified by the dataset
            - 'prediction': the predicted answer label
        - otherwise: list of predicted labels
    references: list of reference labels. Depending on the SuperGLUE subset:
        - for 'record': list of question-answers dictionaries with the following keys:
            - 'idx': index of the question as specified by the dataset
            - 'answers': list of possible answers
        - otherwise: list of reference labels
Returns: depending on the SuperGLUE subset:
    - for 'record':
        - 'exact_match': Exact match between answer and gold answer
        - 'f1': F1 score
    - for 'multirc':
        - 'exact_match': Exact match between answer and gold answer
        - 'f1_m': Per-question macro-F1 score
        - 'f1_a': Average F1 score over all answers
    - for 'axb':
        'matthews_correlation': Matthew Correlation
    - for 'cb':
        - 'accuracy': Accuracy
        - 'f1': F1 score
    - for all others:
        - 'accuracy': Accuracy
Examples:

    >>> super_glue_metric = datasets.load_metric('super_glue', 'copa')  # any of [\"copa\", \"rte\", \"wic\", \"wsc\", \"wsc.fixed\", \"boolq\", \"axg\"]
    >>> predictions = [0, 1]
    >>> references = [0, 1]
    >>> results = super_glue_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'accuracy': 1.0}

    >>> super_glue_metric = datasets.load_metric('super_glue', 'cb')
    >>> predictions = [0, 1]
    >>> references = [0, 1]
    >>> results = super_glue_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'accuracy': 1.0, 'f1': 1.0}

    >>> super_glue_metric = datasets.load_metric('super_glue', 'record')
    >>> predictions = [{'idx': {'passage': 0, 'query': 0}, 'prediction_text': 'answer'}]
    >>> references = [{'idx': {'passage': 0, 'query': 0}, 'answers': ['answer', 'another_answer']}]
    >>> results = super_glue_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'exact_match': 1.0, 'f1': 1.0}

    >>> super_glue_metric = datasets.load_metric('super_glue', 'multirc')
    >>> predictions = [{'idx': {'answer': 0, 'paragraph': 0, 'question': 0}, 'prediction': 0}, {'idx': {'answer': 1, 'paragraph': 2, 'question': 3}, 'prediction': 1}]
    >>> references = [0, 1]
    >>> results = super_glue_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'exact_match': 1.0, 'f1_m': 1.0, 'f1_a': 1.0}

    >>> super_glue_metric = datasets.load_metric('super_glue', 'axb')
    >>> references = [0, 1]
    >>> predictions = [0, 1]
    >>> results = super_glue_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'matthews_correlation': 1.0}
"""
def   _UpperCamelCase				(						lowerCAmelCase_  ,						lowerCAmelCase_   )						->List[str]:
   return float((preds == labels).mean()   )
def   _UpperCamelCase				(						lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_="binary"   )						->Union[str, Any]:
   UpperCAmelCase							=     simple_accuracy(lowerCAmelCase_  ,						lowerCAmelCase_   )
   UpperCAmelCase							=     float(fa_score(y_true=lowerCAmelCase_  ,						y_pred=lowerCAmelCase_  ,						average=lowerCAmelCase_   )   )
   return {
       "accuracy": acc,
       "f1": fa,
   }
def   _UpperCamelCase				(						lowerCAmelCase_  ,						lowerCAmelCase_   )						->List[Any]:
   UpperCAmelCase							=     {}
   for id_pred, label in zip(lowerCAmelCase_  ,						lowerCAmelCase_   ):
      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(*lowerCAmelCase_   )
      UpperCAmelCase							=     fa_score(y_true=lowerCAmelCase_  ,						y_pred=lowerCAmelCase_  ,						average="""macro"""   )
      fas.append(lowerCAmelCase_   )
      UpperCAmelCase							=     int(sum(pred == label for pred, label in preds_labels   ) == len(lowerCAmelCase_   )   )
      ems.append(lowerCAmelCase_   )
   UpperCAmelCase							=     float(sum(lowerCAmelCase_   ) / len(lowerCAmelCase_   )   )
   UpperCAmelCase							=     sum(lowerCAmelCase_   ) / len(lowerCAmelCase_   )
   UpperCAmelCase							=     float(fa_score(y_true=lowerCAmelCase_  ,						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   __lowercase				(  datasets.Metric  ):





      def        _lowercase				(		self						:					int       )    ->   Any:






         """simple docstring"""







         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						:					Optional[Any]       )    ->   Any:






         """simple docstring"""







         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						:					Dict					,       __lowerCamelCase						:					int					,       __lowerCamelCase						:					Optional[int]       )    ->   List[Any]:






         """simple docstring"""







         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\",]"""       )



