






import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
	from .features import FeatureType


UpperCAmelCase_  ,				UpperCAmelCase_  ,				UpperCAmelCase_ :						str      =      False, False, False

@dataclass
class     _SCREAMING_SNAKE_CASE       :
	snake_case__							:     Optional[int]     	=  None
	snake_case__							:     bool     	=  True
	snake_case__							:     bool     	=  True
	snake_case__							:     Optional[str]     	=  None
	# Automatically constructed
	snake_case__							:     ClassVar[str]     	=  "dict"
	snake_case__							:     ClassVar[Any]     	=  pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}      )
	snake_case__							:     str     	=  field(default="""Audio"""		,   init=_a		,   repr=_a      )





	def __call__(		self				:					Optional[int]	):
		return self.pa_type





	def 						_A (		self				:					List[Any] ,  __lowerCamelCase				:					Union[str, bytes, dict]	):
		try:
			import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
		except ImportError as err:
			raise ImportError("""To support encoding audio data, please install 'soundfile'."""	) from err
		if isinstance(__lowerCamelCase ,  __lowerCamelCase	):
			return {"bytes": None, "path": value}
		elif isinstance(__lowerCamelCase ,  __lowerCamelCase	):
			return {"bytes": value, "path": None}
		elif "array" in value:
			# convert the audio array to wav bytes
			UpperCamelCase     :Dict          =	BytesIO()
			sf.write(__lowerCamelCase ,  value["""array"""] ,  value["""sampling_rate"""] ,  format="""wav"""	)
			return {"bytes": buffer.getvalue(), "path": None}
		elif value.get("""path"""	) is not None and os.path.isfile(value["""path"""]	):
			# we set "bytes": None to not duplicate the data if they're already available locally
			if value["path"].endswith("""pcm"""	):
				# "PCM" only has raw audio bytes
				if value.get("""sampling_rate"""	) is None:
					# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
					raise KeyError("""To use PCM files, please specify a 'sampling_rate' in Audio object"""	)
				if value.get("""bytes"""	):
					# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
					UpperCamelCase     :List[str]          =	np.frombuffer(value["""bytes"""] ,  dtype=np.intaa	).astype(np.floataa	) / 32_767
				else:
					UpperCamelCase     :Any          =	np.memmap(value["""path"""] ,  dtype="""h""" ,  mode="""r"""	).astype(np.floataa	) / 32_767

				UpperCamelCase     :Union[str, Any]          =	BytesIO(bytes()	)
				sf.write(__lowerCamelCase ,  __lowerCamelCase ,  value["""sampling_rate"""] ,  format="""wav"""	)
				return {"bytes": buffer.getvalue(), "path": None}
			else:
				return {"bytes": None, "path": value.get("""path"""	)}
		elif value.get("""bytes"""	) is not None or value.get("""path"""	) is not None:
			# store the audio bytes, and path is used to infer the audio format using the file extension
			return {"bytes": value.get("""bytes"""	), "path": value.get("""path"""	)}
		else:
			raise ValueError(
			    F"""An audio sample should have one of 'path' or 'bytes' but they are missing or None in {value}."""	)





	def 						_A (		self				:					List[str] ,  __lowerCamelCase				:					dict ,  __lowerCamelCase				:					Optional[Dict[str, Union[str, bool, None]]] = None	):
		if not self.decode:
			raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""	)

		UpperCamelCase     ,      UpperCamelCase     :int          =	(value["""path"""], BytesIO(value["""bytes"""]	)) if value["""bytes"""] is not None else (value["""path"""], None)
		if path is None and file is None:
			raise ValueError(F"""An audio sample should have one of 'path' or 'bytes' but both are None in {value}."""	)

		try:
			import librosa
			import soundfile as sf
		except ImportError as err:
			raise ImportError("""To support decoding audio files, please install 'librosa' and 'soundfile'."""	) from err

		UpperCamelCase     :int          =	xsplitext(__lowerCamelCase	)[1][1:].lower() if path is not None else None
		if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
			raise RuntimeError(
			    """Decoding 'opus' files requires system library 'libsndfile'>=1.0.31, """
			    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """	)
		elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
			raise RuntimeError(
			    """Decoding 'mp3' files requires system library 'libsndfile'>=1.1.0, """
			    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """	)

		if file is None:
			UpperCamelCase     :int          =	token_per_repo_id or {}
			UpperCamelCase     :Any          =	path.split("""::"""	)[-1]
			try:
				UpperCamelCase     :List[str]          =	string_to_dict(__lowerCamelCase ,  config.HUB_DATASETS_URL	)["""repo_id"""]
				UpperCamelCase     :Union[str, Any]          =	token_per_repo_id[repo_id]
			except (ValueError, KeyError):
				UpperCamelCase     :Tuple          =	None

			with xopen(__lowerCamelCase ,  """rb""" ,  use_auth_token=__lowerCamelCase	) as f:
				UpperCamelCase     ,      UpperCamelCase     :str          =	sf.read(__lowerCamelCase	)

		else:
			UpperCamelCase     ,      UpperCamelCase     :Dict          =	sf.read(__lowerCamelCase	)

		UpperCamelCase     :str          =	array.T
		if self.mono:
			UpperCamelCase     :Dict          =	librosa.to_mono(__lowerCamelCase	)
		if self.sampling_rate and self.sampling_rate != sampling_rate:
			UpperCamelCase     :Optional[int]          =	librosa.resample(__lowerCamelCase ,  orig_sr=__lowerCamelCase ,  target_sr=self.sampling_rate	)
			UpperCamelCase     :Union[str, Any]          =	self.sampling_rate

		return {"path": path, "array": array, "sampling_rate": sampling_rate}





	def 						_A (		self				:					int	):
		from .features import Value

		if self.decode:
			raise ValueError("""Cannot flatten a decoded Audio feature."""	)
		return {
		    "bytes": Value("""binary"""	),
		    "path": Value("""string"""	),
		}





	def 						_A (		self				:					Union[str, Any] ,  __lowerCamelCase				:					Union[pa.StringArray, pa.StructArray]	):
		if pa.types.is_string(storage.type	):
			UpperCamelCase     :Optional[Any]          =	pa.array([None] * len(__lowerCamelCase	) ,  type=pa.binary()	)
			UpperCamelCase     :List[str]          =	pa.StructArray.from_arrays([bytes_array, storage] ,  ["""bytes""", """path"""] ,  mask=storage.is_null()	)
		elif pa.types.is_binary(storage.type	):
			UpperCamelCase     :List[str]          =	pa.array([None] * len(__lowerCamelCase	) ,  type=pa.string()	)
			UpperCamelCase     :Optional[int]          =	pa.StructArray.from_arrays([storage, path_array] ,  ["""bytes""", """path"""] ,  mask=storage.is_null()	)
		elif pa.types.is_struct(storage.type	) and storage.type.get_all_field_indices("""array"""	):
			UpperCamelCase     :Optional[Any]          =	pa.array([Audio().encode_example(__lowerCamelCase	) if x is not None else None for x in storage.to_pylist()]	)
		elif pa.types.is_struct(storage.type	):
			if storage.type.get_field_index("""bytes"""	) >= 0:
				UpperCamelCase     :int          =	storage.field("""bytes"""	)
			else:
				UpperCamelCase     :int          =	pa.array([None] * len(__lowerCamelCase	) ,  type=pa.binary()	)
			if storage.type.get_field_index("""path"""	) >= 0:
				UpperCamelCase     :Tuple          =	storage.field("""path"""	)
			else:
				UpperCamelCase     :List[Any]          =	pa.array([None] * len(__lowerCamelCase	) ,  type=pa.string()	)
			UpperCamelCase     :Optional[Any]          =	pa.StructArray.from_arrays([bytes_array, path_array] ,  ["""bytes""", """path"""] ,  mask=storage.is_null()	)
		return array_cast(__lowerCamelCase ,  self.pa_type	)





	def 						_A (		self				:					int ,  __lowerCamelCase				:					pa.StructArray	):

		@no_op_if_value_is_null
		def path_to_bytes(__lowerCamelCase				:					str	):
			with xopen(__lowerCamelCase ,  """rb"""	) as f:
				UpperCamelCase     :List[str]          =	f.read()
			return bytes_

		UpperCamelCase     :Optional[int]          =	pa.array(
		    [
		        (path_to_bytes(x["""path"""]	) if x["""bytes"""] is None else x["""bytes"""]) if x is not None else None
		        for x in storage.to_pylist()
		    ] ,  type=pa.binary() ,  )
		UpperCamelCase     :Optional[Any]          =	pa.array(
		    [os.path.basename(__lowerCamelCase	) if path is not None else None for path in storage.field("""path"""	).to_pylist()] ,  type=pa.string() ,  )
		UpperCamelCase     :Tuple          =	pa.StructArray.from_arrays([bytes_array, path_array] ,  ["""bytes""", """path"""] ,  mask=bytes_array.is_null()	)
		return array_cast(__lowerCamelCase ,  self.pa_type	)








from sklearn.metrics import fa_score, matthews_corrcoef

import datasets

from .record_evaluation import evaluate as evaluate_record


UpperCAmelCase_ :						int      =      '''\
@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}
}
'''

UpperCAmelCase_ :						Optional[Any]      =      '''\
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.
'''

UpperCAmelCase_ :						int      =      '''
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    SCREAMING_SNAKE_CASE_							(							__magic_name__					:  Optional[Any]	,			__magic_name__					:  Optional[Any]					)		->			Union[str, Any]:






	"""simple docstring"""



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

def    SCREAMING_SNAKE_CASE_							(							__magic_name__					:  Dict	,			__magic_name__					:  int	,			__magic_name__					:  Any="binary"					)		->			Dict:






	"""simple docstring"""



	UpperCamelCase     :List[str]          =	simple_accuracy(__magic_name__	,			__magic_name__					)
	UpperCamelCase     :Dict          =	float(fa_score(y_true=__magic_name__	,			y_pred=__magic_name__	,			average=__magic_name__					)					)
	return {
	    "accuracy": acc,
	    "f1": fa,
	}

def    SCREAMING_SNAKE_CASE_							(							__magic_name__					:  List[Any]	,			__magic_name__					:  Optional[Any]					)		->			Optional[Any]:






	"""simple docstring"""



	UpperCamelCase     :Optional[Any]          =	{}
	for id_pred, label in zip(__magic_name__	,			__magic_name__					):
		UpperCamelCase     :str          =	f"""{id_pred['idx']['paragraph']}-{id_pred['idx']['question']}"""
		UpperCamelCase     :Union[str, Any]          =	id_pred["""prediction"""]
		if question_id in question_map:
			question_map[question_id].append((pred, label)					)
		else:
			UpperCamelCase     :Dict          =	[(pred, label)]
	UpperCamelCase     ,      UpperCamelCase     :Optional[int]          =	[], []
	for question, preds_labels in question_map.items():
		UpperCamelCase     ,      UpperCamelCase     :Optional[Any]          =	zip(*__magic_name__					)
		UpperCamelCase     :Optional[int]          =	fa_score(y_true=__magic_name__	,			y_pred=__magic_name__	,			average="""macro"""					)
		fas.append(__magic_name__					)
		UpperCamelCase     :int          =	int(sum(pred == label for pred, label in preds_labels					) == len(__magic_name__					)					)
		ems.append(__magic_name__					)
	UpperCamelCase     :Optional[int]          =	float(sum(__magic_name__					) / len(__magic_name__					)					)
	UpperCamelCase     :str          =	sum(__magic_name__					) / len(__magic_name__					)
	UpperCamelCase     :Tuple          =	float(fa_score(y_true=__magic_name__	,			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     _SCREAMING_SNAKE_CASE       (			datasets.Metric      ):





	def 						_A (		self				:					str	):
		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 						_A (		self				:					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 						_A (		self				:					List[str] ,  __lowerCamelCase				:					Tuple ,  __lowerCamelCase				:					str	):
		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     :Optional[Any]          =	[
			    {
			        """qas""": [
			            {"""id""": ref["""idx"""]["""query"""], """answers""": [{"""text""": ans} for ans in ref["""answers"""]]}
			            for ref in references
			        ]
			    }
			]
			UpperCamelCase     :Tuple          =	{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\",]"""	)
