



from typing import List

from packaging import version
from sklearn.metrics import fa_score

import datasets
from datasets.config import PY_VERSION


if PY_VERSION < version.parse("""3.8"""):
      import importlib_metadata
else:
      import importlib.metadata as importlib_metadata


# TODO(Patrick/Anton)
_lowerCamelCase         =	"""\
"""

_lowerCamelCase         =	"""\
    XTREME-S is a benchmark to evaluate universal cross-lingual speech representations in many languages.
    XTREME-S covers four task families: speech recognition, classification, speech-to-text translation and retrieval.
"""

_lowerCamelCase         =	"""
Compute XTREME-S evaluation metric associated to each XTREME-S dataset.
Args:
    predictions: list of predictions to score.
        Each translation should be tokenized into a list of tokens.
    references: list of lists of references for each translation.
        Each reference should be tokenized into a list of tokens.
    bleu_kwargs: optional dict of keywords to be passed when computing 'bleu'.
        Keywords include Dict can be one of 'smooth_method', 'smooth_value', 'force', 'lowercase',
        'tokenize', 'use_effective_order'.
    wer_kwargs: optional dict of keywords to be passed when computing 'wer' and 'cer'.
        Keywords include 'concatenate_texts'.
Returns: depending on the XTREME-S task, one or several of:
    \"accuracy\": Accuracy - for 'fleurs-lang_id', 'minds14'
    \"f1\": F1 score - for 'minds14'
    \"wer\": Word error rate - for 'mls', 'fleurs-asr', 'voxpopuli', 'babel'
    \"cer\": Character error rate - for 'mls', 'fleurs-asr', 'voxpopuli', 'babel'
    \"bleu\": BLEU score according to the `sacrebleu` metric - for 'covost2'
Examples:

    >>> xtreme_s_metric = datasets.load_metric('xtreme_s', 'mls')  # 'mls', 'voxpopuli', 'fleurs-asr' or 'babel'
    >>> references = [\"it is sunny here\", \"paper and pen are essentials\"]
    >>> predictions = [\"it's sunny\", \"paper pen are essential\"]
    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)
    >>> print({k: round(v, 2) for k, v in results.items()})
    {'wer': 0.56, 'cer': 0.27}

    >>> xtreme_s_metric = datasets.load_metric('xtreme_s', 'covost2')
    >>> references = [\"bonjour paris\", \"il est necessaire de faire du sport de temps en temp\"]
    >>> predictions = [\"bonjour paris\", \"il est important de faire du sport souvent\"]
    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)
    >>> print({k: round(v, 2) for k, v in results.items()})
    {'bleu': 31.65}

    >>> xtreme_s_metric = datasets.load_metric('xtreme_s', 'fleurs-lang_id')
    >>> references = [0, 1, 0, 0, 1]
    >>> predictions = [0, 1, 1, 0, 0]
    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)
    >>> print({k: round(v, 2) for k, v in results.items()})
    {'accuracy': 0.6}

    >>> xtreme_s_metric = datasets.load_metric('xtreme_s', 'minds14')
    >>> references = [0, 1, 0, 0, 1]
    >>> predictions = [0, 1, 1, 0, 0]
    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)
    >>> print({k: round(v, 2) for k, v in results.items()})
    {'f1': 0.58, 'accuracy': 0.6}
"""

_lowerCamelCase         =	["""fleurs-asr""", """mls""", """voxpopuli""", """babel""", """covost2""", """fleurs-lang_id""", """minds14"""]
_lowerCamelCase         =	""""""

try:
      from jiwer import transforms as tr

      _lowerCamelCase         =	True
except ImportError:
      _lowerCamelCase         =	False

if _jiwer_available and version.parse(importlib_metadata.version("""jiwer""")) < version.parse("""2.3.0"""):

      class    _SCREAMING_SNAKE_CASE						(tr.AbstractTransform			):





           def __init__(							self			:				Tuple  ,	UpperCamelCase			:				str = " "					)->List[str]:
            __SCREAMING_SNAKE_CASE							:     List[Any]							=						sentence_delimiter





           def 						__snake_case     (							self			:				Union[str, Any]  ,	UpperCamelCase			:				str					)->Any:
            return list(UpperCamelCase					)






           def 						__snake_case     (							self			:				str  ,	UpperCamelCase			:				List[str]					)->Tuple:
            __SCREAMING_SNAKE_CASE							:     int							=						[]
            for sent_idx, sentence in enumerate(UpperCamelCase					):
             chars.extend(self.process_string(UpperCamelCase					)					)
             if self.sentence_delimiter is not None and self.sentence_delimiter != "" and sent_idx < len(UpperCamelCase					) - 1:
              chars.append(self.sentence_delimiter					)
            return chars






      _lowerCamelCase         =	tr.Compose(
          [tr.RemoveMultipleSpaces(), tr.Strip(), SentencesToListOfCharacters(SENTENCE_DELIMITER)]
      )
elif _jiwer_available:
      _lowerCamelCase         =	tr.Compose(
          [
              tr.RemoveMultipleSpaces(),
              tr.Strip(),
              tr.ReduceToSingleSentence(SENTENCE_DELIMITER),
              tr.ReduceToListOfListOfChars(),
          ]
      )
else:
      _lowerCamelCase         =	None






def 						_lowerCAmelCase    (					__lowerCamelCase :  Optional[Any]		,						__lowerCamelCase :  Union[str, Any]  ):



 """simple docstring"""





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






def 						_lowerCAmelCase    (					__lowerCamelCase :  Dict		,						__lowerCamelCase :  str  ):



 """simple docstring"""





 return {
     "f1": float(fa_score(y_true=__lowerCamelCase		,						y_pred=__lowerCamelCase		,						average="macro"  )  ),
     "accuracy": simple_accuracy(__lowerCamelCase		,						__lowerCamelCase  ),
 }






def 						_lowerCAmelCase    (					__lowerCamelCase :  str		,						__lowerCamelCase :  List[str]		,						__lowerCamelCase :  Optional[int]="exp"		,						__lowerCamelCase :  Optional[Any]=None		,						__lowerCamelCase :  Tuple=False		,						__lowerCamelCase :  Dict=False		,						__lowerCamelCase :  Dict=None		,						__lowerCamelCase :  Optional[Any]=False		,						):



 """simple docstring"""





 __SCREAMING_SNAKE_CASE							:     Tuple							=						[[label] for label in labels]
 __SCREAMING_SNAKE_CASE							:     List[Any]							=						list(__lowerCamelCase  )
 try:
  import sacrebleu as scb
 except ImportError:
  raise ValueError(
      "sacrebleu has to be installed in order to apply the bleu metric for covost2."
      "You can install it via `pip install sacrebleu`."  )

 if version.parse(scb.__version__  ) < version.parse("1.4.12"  ):
  raise ImportWarning(
      "To use `sacrebleu`, the module `sacrebleu>=1.4.12` is required, and the current version of `sacrebleu` doesn't match this condition.\n"
      "You can install it with `pip install \"sacrebleu>=1.4.12\"`."  )

 __SCREAMING_SNAKE_CASE							:     Optional[int]							=						len(labels[0]  )
 if any(len(__lowerCamelCase  ) != references_per_prediction for refs in labels  ):
  raise ValueError("Sacrebleu requires the same number of references for each prediction"  )
 __SCREAMING_SNAKE_CASE							:     Optional[Any]							=						[[refs[i] for refs in labels] for i in range(__lowerCamelCase  )]
 __SCREAMING_SNAKE_CASE							:     List[Any]							=						scb.corpus_bleu(
     __lowerCamelCase		,						__lowerCamelCase		,						smooth_method=__lowerCamelCase		,						smooth_value=__lowerCamelCase		,						force=__lowerCamelCase		,						lowercase=__lowerCamelCase		,						use_effective_order=__lowerCamelCase		,						**({"tokenize": tokenize} if tokenize else {})		,						)
 return {"bleu": output.score}






def 						_lowerCAmelCase    (					__lowerCamelCase :  List[str]		,						__lowerCamelCase :  List[str]		,						__lowerCamelCase :  Union[str, Any]		,						__lowerCamelCase :  int  ):



 """simple docstring"""





 try:
  from jiwer import compute_measures
 except ImportError:
  raise ValueError(
      F"""jiwer has to be installed in order to apply the wer metric for {config_name}."""
      "You can install it via `pip install jiwer`."  )

 if concatenate_texts:
  __SCREAMING_SNAKE_CASE							:     Dict							=						compute_measures(__lowerCamelCase		,						__lowerCamelCase  )["wer"]

  __SCREAMING_SNAKE_CASE							:     List[Any]							=						compute_measures(__lowerCamelCase		,						__lowerCamelCase		,						truth_transform=__lowerCamelCase		,						hypothesis_transform=__lowerCamelCase  )["wer"]
  return {"wer": wer, "cer": cer}
 else:

  def compute_score(__lowerCamelCase :  Any		,						__lowerCamelCase :  Optional[int]		,						__lowerCamelCase :  List[Any]="wer"  ):
   __SCREAMING_SNAKE_CASE							:     List[Any]							=						0
   __SCREAMING_SNAKE_CASE							:     Optional[int]							=						0
   for prediction, reference in zip(__lowerCamelCase		,						__lowerCamelCase  ):
    if score_type == "wer":
     __SCREAMING_SNAKE_CASE							:     List[str]							=						compute_measures(__lowerCamelCase		,						__lowerCamelCase  )
    elif score_type == "cer":
     __SCREAMING_SNAKE_CASE							:     Optional[Any]							=						compute_measures(
         __lowerCamelCase		,						__lowerCamelCase		,						truth_transform=__lowerCamelCase		,						hypothesis_transform=__lowerCamelCase  )
    incorrect += measures["substitutions"] + measures["deletions"] + measures["insertions"]
    total += measures["substitutions"] + measures["deletions"] + measures["hits"]
   return incorrect / total

  return {"wer": compute_score(__lowerCamelCase		,						__lowerCamelCase		,						"wer"  ), "cer": compute_score(__lowerCamelCase		,						__lowerCamelCase		,						"cer"  )}






@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION      ,     _KWARGS_DESCRIPTION			)
class    _SCREAMING_SNAKE_CASE						(datasets.Metric			):





     def 						__snake_case     (							self			:				Optional[Any]					)->Tuple:
      if self.config_name not in _CONFIG_NAMES:
       raise KeyError(F"""You should supply a configuration name selected in {_CONFIG_NAMES}"""					)

      __SCREAMING_SNAKE_CASE							:     Optional[Any]							=						"int64" if self.config_name in ["fleurs-lang_id", "minds14"] else "string"

      return datasets.MetricInfo(
          description=_DESCRIPTION  ,	citation=_CITATION  ,	inputs_description=_KWARGS_DESCRIPTION  ,	features=datasets.Features(
              {"predictions": datasets.Value(UpperCamelCase					), "references": datasets.Value(UpperCamelCase					)}					)  ,	codebase_urls=[]  ,	reference_urls=[]  ,	format="numpy"  ,	)






     def 						__snake_case     (							self			:				List[Any]  ,	UpperCamelCase			:				Union[str, Any]  ,	UpperCamelCase			:				List[str]  ,	UpperCamelCase			:				Union[str, Any]=None  ,	UpperCamelCase			:				Dict=None					)->Optional[int]:
      __SCREAMING_SNAKE_CASE							:     str							=						bleu_kwargs if bleu_kwargs is not None else {}
      __SCREAMING_SNAKE_CASE							:     Any							=						wer_kwargs if wer_kwargs is not None else {}

      if self.config_name == "fleurs-lang_id":
       return {"accuracy": simple_accuracy(UpperCamelCase  ,	UpperCamelCase					)}
      elif self.config_name == "minds14":
       return fa_and_simple_accuracy(UpperCamelCase  ,	UpperCamelCase					)
      elif self.config_name == "covost2":
       __SCREAMING_SNAKE_CASE							:     Optional[Any]							=						bleu_kwargs.pop("smooth_method"  ,	"exp"					)
       __SCREAMING_SNAKE_CASE							:     List[str]							=						bleu_kwargs.pop("smooth_value"  ,	UpperCamelCase					)
       __SCREAMING_SNAKE_CASE							:     str							=						bleu_kwargs.pop("force"  ,	UpperCamelCase					)
       __SCREAMING_SNAKE_CASE							:     Tuple							=						bleu_kwargs.pop("lowercase"  ,	UpperCamelCase					)
       __SCREAMING_SNAKE_CASE							:     Optional[Any]							=						bleu_kwargs.pop("tokenize"  ,	UpperCamelCase					)
       __SCREAMING_SNAKE_CASE							:     Dict							=						bleu_kwargs.pop("use_effective_order"  ,	UpperCamelCase					)
       return bleu(
           preds=UpperCamelCase  ,	labels=UpperCamelCase  ,	smooth_method=UpperCamelCase  ,	smooth_value=UpperCamelCase  ,	force=UpperCamelCase  ,	lowercase=UpperCamelCase  ,	tokenize=UpperCamelCase  ,	use_effective_order=UpperCamelCase  ,	)
      elif self.config_name in ["fleurs-asr", "mls", "voxpopuli", "babel"]:
       __SCREAMING_SNAKE_CASE							:     Union[str, Any]							=						wer_kwargs.pop("concatenate_texts"  ,	UpperCamelCase					)
       return wer_and_cer(UpperCamelCase  ,	UpperCamelCase  ,	UpperCamelCase  ,	self.config_name					)
      else:
       raise KeyError(F"""You should supply a configuration name selected in {_CONFIG_NAMES}"""					)












'''simple docstring'''






import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


_lowerCAmelCase       =   namedtuple('''ExampleNamedTuple''', '''a b c''')



class    A						(					unittest.TestCase						):




				'''simple docstring'''





				def 			a_			(self     )      ->			Tuple:
							__UpperCamelCase :     Optional[int]										=  torch.randn(5					,    2     )
							__UpperCamelCase :     List[str]										=  torch.device("cuda"     ) if torch.cuda.is_available() else torch.device("cpu"     )

							__UpperCamelCase :     Optional[Any]										=  send_to_device(_UpperCAmelCase					,    _UpperCAmelCase     )
							self.assertTrue(torch.equal(resulta.cpu()					,    _UpperCAmelCase     )     )

							__UpperCamelCase :     Any										=  send_to_device((tensor, [tensor, tensor], 1)					,    _UpperCAmelCase     )
							self.assertIsInstance(_UpperCAmelCase					,    _UpperCAmelCase     )
							self.assertTrue(torch.equal(resulta[0].cpu()					,    _UpperCAmelCase     )     )
							self.assertIsInstance(resulta[1]					,    _UpperCAmelCase     )
							self.assertTrue(torch.equal(resulta[1][0].cpu()					,    _UpperCAmelCase     )     )
							self.assertTrue(torch.equal(resulta[1][1].cpu()					,    _UpperCAmelCase     )     )
							self.assertEqual(resulta[2]					,    1     )

							__UpperCamelCase :     Optional[int]										=  send_to_device({"a": tensor, "b": [tensor, tensor], "c": 1}					,    _UpperCAmelCase     )
							self.assertIsInstance(_UpperCAmelCase					,    _UpperCAmelCase     )
							self.assertTrue(torch.equal(resulta["a"].cpu()					,    _UpperCAmelCase     )     )
							self.assertIsInstance(resulta["b"]					,    _UpperCAmelCase     )
							self.assertTrue(torch.equal(resulta["b"][0].cpu()					,    _UpperCAmelCase     )     )
							self.assertTrue(torch.equal(resulta["b"][1].cpu()					,    _UpperCAmelCase     )     )
							self.assertEqual(resulta["c"]					,    1     )

							__UpperCamelCase :     Optional[Any]										=  send_to_device(ExampleNamedTuple(a=_UpperCAmelCase					,    b=[tensor, tensor]					,    c=1     )					,    _UpperCAmelCase     )
							self.assertIsInstance(_UpperCAmelCase					,    _UpperCAmelCase     )
							self.assertTrue(torch.equal(resulta.a.cpu()					,    _UpperCAmelCase     )     )
							self.assertIsInstance(resulta.b					,    _UpperCAmelCase     )
							self.assertTrue(torch.equal(resulta.b[0].cpu()					,    _UpperCAmelCase     )     )
							self.assertTrue(torch.equal(resulta.b[1].cpu()					,    _UpperCAmelCase     )     )
							self.assertEqual(resulta.c					,    1     )

							__UpperCamelCase :     Dict										=  send_to_device(UserDict({"a": tensor, "b": [tensor, tensor], "c": 1}     )					,    _UpperCAmelCase     )
							self.assertIsInstance(_UpperCAmelCase					,    _UpperCAmelCase     )
							self.assertTrue(torch.equal(resulta["a"].cpu()					,    _UpperCAmelCase     )     )
							self.assertIsInstance(resulta["b"]					,    _UpperCAmelCase     )
							self.assertTrue(torch.equal(resulta["b"][0].cpu()					,    _UpperCAmelCase     )     )
							self.assertTrue(torch.equal(resulta["b"][1].cpu()					,    _UpperCAmelCase     )     )
							self.assertEqual(resulta["c"]					,    1     )
				def 			a_			(self     )      ->			Optional[Any]:
							with self.assertRaises(_UpperCAmelCase     ) as cm:
										__UpperCamelCase :     int										=  recursively_apply(torch.tensor					,    (torch.tensor(1     ), 1)					,    error_on_other_type=_UpperCAmelCase     )
							self.assertEqual(
							    str(cm.exception     )					,    "Unsupported types (<class 'int'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed."					,    )
				def 			a_			(self     )      ->			Union[str, Any]:
							__UpperCamelCase :     Dict										=  torch.tensor([1, 2, 3, 4, 5]     )
							self.assertEqual(listify(_UpperCAmelCase     )					,    [1, 2, 3, 4, 5]     )

							__UpperCamelCase :     int										=  torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]     )
							self.assertEqual(listify(_UpperCAmelCase     )					,    [[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]]     )

							__UpperCamelCase :     Dict										=  torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]     )
							self.assertEqual(
							    listify(_UpperCAmelCase     )					,    [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 1_0]], [[1_1, 1_2, 1_3, 1_4, 1_5], [1_6, 1_7, 1_8, 1_9, 2_0]]]     )
				def 			a_			(self     )      ->			Optional[Any]:
							with patch_environment(aa=1					,    BB=2     ):
										self.assertEqual(os.environ.get("AA"     )					,    "1"     )
										self.assertEqual(os.environ.get("BB"     )					,    "2"     )

							self.assertNotIn("AA"					,    os.environ     )
							self.assertNotIn("BB"					,    os.environ     )
				def 			a_			(self     )      ->			Dict:
							__UpperCamelCase :     Tuple										=  RegressionModel()
							__UpperCamelCase :     List[str]										=  model.forward
							__UpperCamelCase :     Any										=  convert_outputs_to_fpaa(model.forward     )
							__UpperCamelCase :     List[str]										=  extract_model_from_parallel(_UpperCAmelCase					,    keep_fpaa_wrapper=_UpperCAmelCase     )
							__UpperCamelCase :     Dict										=  pickle.dumps(_UpperCAmelCase     )
				@require_cuda
				def 			a_			(self     )      ->			str:
							__UpperCamelCase :     List[str]										=  RegressionModel()
							__UpperCamelCase :     List[Any]										=  model.forward
							__UpperCamelCase :     List[str]										=  torch.cuda.amp.autocast(dtype=torch.floataa     )(model.forward     )
							__UpperCamelCase :     str										=  convert_outputs_to_fpaa(model.forward     )
							__UpperCamelCase :     int										=  extract_model_from_parallel(_UpperCAmelCase					,    keep_fpaa_wrapper=_UpperCAmelCase     )
							__UpperCamelCase :     int										=  pickle.dumps(_UpperCAmelCase     )
				@require_cuda
				@require_torch_min_version(version="2.0"     )
				def 			a_			(self     )      ->			Tuple:
							__UpperCamelCase :     List[Any]										=  RegressionModel()
							__UpperCamelCase :     Tuple										=  model.forward
							__UpperCamelCase :     Optional[int]										=  torch.cuda.amp.autocast(dtype=torch.floataa     )(model.forward     )
							__UpperCamelCase :     List[Any]										=  convert_outputs_to_fpaa(model.forward     )
							__UpperCamelCase :     Union[str, Any]										=  torch.compile(model.forward					,    backend="inductor"     )
							__UpperCamelCase :     int										=  torch.randn(4					,    1_0     ).cuda()
							__UpperCamelCase :     int										=  model(_UpperCAmelCase     )
				def 			a_			(self     )      ->			Union[str, Any]:
							__UpperCamelCase :     List[Any]										=  RegressionModel()
							# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
							__UpperCamelCase :     Optional[Any]										=  torch.nn.parallel.DataParallel(_UpperCAmelCase     )
							__UpperCamelCase :     Union[str, Any]										=  extract_model_from_parallel(_UpperCAmelCase     )

							self.assertEqual(_UpperCAmelCase					,    _UpperCAmelCase     )
				@require_torch_min_version(version="2.0"     )
				def 			a_			(self     )      ->			Union[str, Any]:
							__UpperCamelCase :     int										=  RegressionModel()
							__UpperCamelCase :     List[str]										=  torch.compile(_UpperCAmelCase     )

							# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
							__UpperCamelCase :     Union[str, Any]										=  torch.nn.parallel.DataParallel(_UpperCAmelCase     )
							__UpperCamelCase :     List[str]										=  torch.compile(_UpperCAmelCase     )
							__UpperCamelCase :     Optional[int]										=  extract_model_from_parallel(_UpperCAmelCase     )

							self.assertEqual(compiled_model._orig_mod					,    compiled_model_unwrapped._orig_mod     )





				def 			a_			(self     )      ->			str:
							self.assertEqual(find_device([1, "a", torch.tensor([1, 2, 3]     )]     )					,    torch.device("cpu"     )     )
							self.assertEqual(find_device({"a": 1, "b": torch.tensor([1, 2, 3]     )}     )					,    torch.device("cpu"     )     )
							self.assertIsNone(find_device([1, "a"]     )     )



