
"""simple docstring"""




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      :					Tuple   	=       '\\n'

__lowerCAmelCase      :					Optional[int]   	=       '\\n    XTREME-S is a benchmark to evaluate universal cross-lingual speech representations in many languages.\n    XTREME-S covers four task families: speech recognition, classification, speech-to-text translation and retrieval.\n'

__lowerCAmelCase      :					Tuple   	=       '\nCompute XTREME-S evaluation metric associated to each XTREME-S dataset.\nArgs:\n    predictions: list of predictions to score.\n        Each translation should be tokenized into a list of tokens.\n    references: list of lists of references for each translation.\n        Each reference should be tokenized into a list of tokens.\n    bleu_kwargs: optional dict of keywords to be passed when computing \'bleu\'.\n        Keywords include Dict can be one of \'smooth_method\', \'smooth_value\', \'force\', \'lowercase\',\n        \'tokenize\', \'use_effective_order\'.\n    wer_kwargs: optional dict of keywords to be passed when computing \'wer\' and \'cer\'.\n        Keywords include \'concatenate_texts\'.\nReturns: depending on the XTREME-S task, one or several of:\n    "accuracy": Accuracy - for \'fleurs-lang_id\', \'minds14\'\n    "f1": F1 score - for \'minds14\'\n    "wer": Word error rate - for \'mls\', \'fleurs-asr\', \'voxpopuli\', \'babel\'\n    "cer": Character error rate - for \'mls\', \'fleurs-asr\', \'voxpopuli\', \'babel\'\n    "bleu": BLEU score according to the `sacrebleu` metric - for \'covost2\'\nExamples:\n\n    >>> xtreme_s_metric = datasets.load_metric(\'xtreme_s\', \'mls\')  # \'mls\', \'voxpopuli\', \'fleurs-asr\' or \'babel\'\n    >>> references = ["it is sunny here", "paper and pen are essentials"]\n    >>> predictions = ["it\'s sunny", "paper pen are essential"]\n    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)\n    >>> print({k: round(v, 2) for k, v in results.items()})\n    {\'wer\': 0.56, \'cer\': 0.27}\n\n    >>> xtreme_s_metric = datasets.load_metric(\'xtreme_s\', \'covost2\')\n    >>> references = ["bonjour paris", "il est necessaire de faire du sport de temps en temp"]\n    >>> predictions = ["bonjour paris", "il est important de faire du sport souvent"]\n    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)\n    >>> print({k: round(v, 2) for k, v in results.items()})\n    {\'bleu\': 31.65}\n\n    >>> xtreme_s_metric = datasets.load_metric(\'xtreme_s\', \'fleurs-lang_id\')\n    >>> references = [0, 1, 0, 0, 1]\n    >>> predictions = [0, 1, 1, 0, 0]\n    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)\n    >>> print({k: round(v, 2) for k, v in results.items()})\n    {\'accuracy\': 0.6}\n\n    >>> xtreme_s_metric = datasets.load_metric(\'xtreme_s\', \'minds14\')\n    >>> references = [0, 1, 0, 0, 1]\n    >>> predictions = [0, 1, 1, 0, 0]\n    >>> results = xtreme_s_metric.compute(predictions=predictions, references=references)\n    >>> print({k: round(v, 2) for k, v in results.items()})\n    {\'f1\': 0.58, \'accuracy\': 0.6}\n'

__lowerCAmelCase      :					Tuple   	=       ['fleurs-asr', 'mls', 'voxpopuli', 'babel', 'covost2', 'fleurs-lang_id', 'minds14']
__lowerCAmelCase      :					int   	=       ''

try:
    from jiwer import transforms as tr

    __lowerCAmelCase      :					List[Any]   	=       True
except ImportError:
    __lowerCAmelCase      :					Any   	=       False

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




    class 	UpperCAmelCase_      (			tr.AbstractTransform					):


      '''simple docstring'''







      def __init__(			self      :			Optional[Any]      ,							UpperCamelCase__      :			str = " "   )      ->     Dict:







        """simple docstring"""






        __magic_name__    					=	sentence_delimiter
      def 							_lowercase							(			self      :			List[Any]      ,							UpperCamelCase__      :			str   )      ->     List[str]:







        """simple docstring"""






        return list(lowercase_   )





      def 							_lowercase							(			self      :			Any      ,							UpperCamelCase__      :			List[str]   )      ->     Dict:







        """simple docstring"""






        __magic_name__    					=	[]
        for sent_idx, sentence in enumerate(lowercase_   ):
          chars.extend(self.process_string(lowercase_   )   )
          if self.sentence_delimiter is not None and self.sentence_delimiter != "" and sent_idx < len(lowercase_   ) - 1:
            chars.append(self.sentence_delimiter   )
        return chars







    __lowerCAmelCase      :					Optional[Any]   	=       tr.Compose(
        [tr.RemoveMultipleSpaces(), tr.Strip(), SentencesToListOfCharacters(SENTENCE_DELIMITER)]
    )
elif _jiwer_available:
    __lowerCAmelCase      :					List[Any]   	=       tr.Compose(
        [
            tr.RemoveMultipleSpaces(),
            tr.Strip(),
            tr.ReduceToSingleSentence(SENTENCE_DELIMITER),
            tr.ReduceToListOfListOfChars(),
        ]
    )
else:
    __lowerCAmelCase      :					int   	=       None

def a__   (	A_,					A_ ):






  '''simple docstring'''





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

def a__   (	A_,					A_ ):






  '''simple docstring'''





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

def a__   (	A_,					A_,					A_="exp",					A_=None,					A_=False,					A_=False,					A_=None,					A_=False,					):






  '''simple docstring'''





  __magic_name__    					=	[[label] for label in labels]
  __magic_name__    					=	list(SCREAMING_SNAKE_CASE__ )
  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\"`.""" )

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

def a__   (	A_,					A_,					A_,					A_ ):






  '''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:
    __magic_name__    					=	compute_measures(SCREAMING_SNAKE_CASE__,					SCREAMING_SNAKE_CASE__ )['wer']

    __magic_name__    					=	compute_measures(SCREAMING_SNAKE_CASE__,					SCREAMING_SNAKE_CASE__,					truth_transform=SCREAMING_SNAKE_CASE__,					hypothesis_transform=SCREAMING_SNAKE_CASE__ )['wer']
    return {"wer": wer, "cer": cer}
  else:

    def compute_score(A_,					A_,					A_="wer" ):
      __magic_name__    					=	0
      __magic_name__    					=	0
      for prediction, reference in zip(SCREAMING_SNAKE_CASE__,					SCREAMING_SNAKE_CASE__ ):
        if score_type == "wer":
          __magic_name__    					=	compute_measures(SCREAMING_SNAKE_CASE__,					SCREAMING_SNAKE_CASE__ )
        elif score_type == "cer":
          __magic_name__    					=	compute_measures(
              SCREAMING_SNAKE_CASE__,					SCREAMING_SNAKE_CASE__,					truth_transform=SCREAMING_SNAKE_CASE__,					hypothesis_transform=SCREAMING_SNAKE_CASE__ )
        incorrect += measures["substitutions"] + measures["deletions"] + measures["insertions"]
        total += measures["substitutions"] + measures["deletions"] + measures["hits"]
      return incorrect / total

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





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


  '''simple docstring'''







  def 							_lowercase							(			self      :			str   )      ->     Optional[Any]:







    """simple docstring"""






    if self.config_name not in _CONFIG_NAMES:
      raise KeyError(F'''You should supply a configuration name selected in {_CONFIG_NAMES}'''   )

    __magic_name__    					=	'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(lowercase_   ), """references""": datasets.Value(lowercase_   )}   )      ,							codebase_urls=[]      ,							reference_urls=[]      ,							format="""numpy"""      ,							)





  def 							_lowercase							(			self      :			Optional[Any]      ,							UpperCamelCase__      :			Optional[int]      ,							UpperCamelCase__      :			List[str]      ,							UpperCamelCase__      :			Optional[Any]=None      ,							UpperCamelCase__      :			List[Any]=None   )      ->     Dict:







    """simple docstring"""






    __magic_name__    					=	bleu_kwargs if bleu_kwargs is not None else {}
    __magic_name__    					=	wer_kwargs if wer_kwargs is not None else {}

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



import bisect
import copy
from collections import defaultdict

import numpy as np
from torch.utils.data import BatchSampler, Sampler

from utils import logger

def a__   (	A_,					A_ ):






  '''simple docstring'''





  __magic_name__    					=	copy.deepcopy(A_ )
  __magic_name__    					=	sorted(A_ )
  __magic_name__    					=	[bisect.bisect_right(A_,					A_ ) for y in x]
  return quantized

def a__   (	A_,					A_=0 ):






  '''simple docstring'''





  __magic_name__    					=	np.arange(start=3,					stop=A_,					step=4 ).tolist() if k > 0 else [10]
  __magic_name__    					=	_quantize(A_,					A_ )
  # count number of elements per group
  __magic_name__    					=	np.unique(A_,					return_counts=A_ )[1]
  __magic_name__    					=	[0] + bins + [np.inf]
  logger.info("""Using {} as bins for aspect lengths quantization""".format(A_ ) )
  logger.info("""Count of instances per bin: {}""".format(A_ ) )
  return groups





class 	UpperCAmelCase_      (			_A					):


  '''simple docstring'''







  def __init__(			self      :			Dict      ,							UpperCamelCase__      :			Dict      ,							UpperCamelCase__      :			Optional[Any]      ,							UpperCamelCase__      :			Tuple   )      ->     Union[str, Any]:







    """simple docstring"""






    if not isinstance(UpperCamelCase__      ,							UpperCamelCase__   ):
      raise ValueError(
          """sampler should be an instance of torch.utils.data.Sampler, but got sampler={}""".format(UpperCamelCase__   )   )
    __magic_name__    					=	sampler
    __magic_name__    					=	group_ids
    __magic_name__    					=	batch_size
  def __iter__(			self      :			Tuple   )      ->     Union[str, Any]:







    """simple docstring"""






    __magic_name__    					=	defaultdict(UpperCamelCase__   )
    __magic_name__    					=	defaultdict(UpperCamelCase__   )

    __magic_name__    					=	0
    for idx in self.sampler:
      __magic_name__    					=	self.group_ids[idx]
      buffer_per_group[group_id].append(UpperCamelCase__   )
      samples_per_group[group_id].append(UpperCamelCase__   )
      if len(buffer_per_group[group_id]   ) == self.batch_size:
        yield buffer_per_group[group_id]  # TODO
        num_batches += 1
        del buffer_per_group[group_id]
      assert len(buffer_per_group[group_id]   ) < self.batch_size

    # now we have run out of elements that satisfy
    # the group criteria, let's return the remaining
    # elements so that the size of the sampler is
    # deterministic
    __magic_name__    					=	len(self   )
    __magic_name__    					=	expected_num_batches - num_batches
    if num_remaining > 0:
      # for the remaining batches, group the batches by similar lengths
      __magic_name__    					=	[]
      for group_id, idxs in sorted(buffer_per_group.items()      ,							key=lambda UpperCamelCase__   : x[0]   ):
        batch_idx.extend(UpperCamelCase__   )
        if len(UpperCamelCase__   ) >= self.batch_size:
          yield batch_idx[: self.batch_size]
          __magic_name__    					=	batch_idx[self.batch_size :]
          num_remaining -= 1
      if len(UpperCamelCase__   ) > 0:
        yield batch_idx
        num_remaining -= 1
    assert num_remaining == 0





  def __len__(			self      :			List[Any]   )      ->     Any:







    """simple docstring"""






    return (len(self.sampler   ) + self.batch_size - 1) // self.batch_size

