




"""simple docstring"""






import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


lowercase__  		=		logging.get_logger(__name__)






class lowerCAmelCase__			(  lowercase		):






			'''simple docstring'''



			lowerCamelCase__   =    ["""input_features"""]
			def __init__(						self						,   lowercase=80						,   lowercase=16000						,   lowercase=160						,   lowercase=30						,   lowercase=400						,   lowercase=0.0						,   lowercase=False						,   **lowercase						,   ):
							super().__init__(
							    feature_size=lowercase						,   sampling_rate=lowercase						,   padding_value=lowercase						,   return_attention_mask=lowercase						,   **lowercase						,   )
							_lowerCamelCase  :    List[Any]					=					n_fft
							_lowerCamelCase  :    Optional[Any]					=					hop_length
							_lowerCamelCase  :    List[str]					=					chunk_length
							_lowerCamelCase  :    Optional[Any]					=					chunk_length * sampling_rate
							_lowerCamelCase  :    Any					=					self.n_samples // hop_length
							_lowerCamelCase  :    Union[str, Any]					=					sampling_rate
							_lowerCamelCase  :    Union[str, Any]					=					mel_filter_bank(
							    num_frequency_bins=1 + n_fft // 2						,   num_mel_filters=lowercase						,   min_frequency=0.0						,   max_frequency=80_00.0						,   sampling_rate=lowercase						,   norm='slaney'						,   mel_scale='slaney'						,   )
			def       A_					(						self						,   lowercase						):
							_lowerCamelCase  :    Union[str, Any]					=					spectrogram(
							    lowercase						,   window_function(self.n_fft						,   'hann'						)						,   frame_length=self.n_fft						,   hop_length=self.hop_length						,   power=2.0						,   mel_filters=self.mel_filters						,   log_mel='log10'						,   )
							_lowerCamelCase  :    Optional[Any]					=					log_spec[:, :-1]
							_lowerCamelCase  :    List[Any]					=					np.maximum(lowercase						,   log_spec.max() - 8.0						)
							_lowerCamelCase  :    List[Any]					=					(log_spec + 4.0) / 4.0
							return log_spec
			@staticmethod
			# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
			def       A_					(						lowercase						,   lowercase						,   lowercase = 0.0						):
							if attention_mask is not None:
											_lowerCamelCase  :    Union[str, Any]					=					np.array(lowercase						,   np.intaa						)
											_lowerCamelCase  :    Tuple					=					[]

											for vector, length in zip(lowercase						,   attention_mask.sum(-1						)						):
															_lowerCamelCase  :    Dict					=					(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7						)
															if length < normed_slice.shape[0]:
																			_lowerCamelCase  :    List[str]					=					padding_value

															normed_input_values.append(lowercase						)
							else:
											_lowerCamelCase  :    Dict					=					[(x - x.mean()) / np.sqrt(x.var() + 1E-7						) for x in input_values]

							return normed_input_values
			def __call__(						self						,   lowercase						,   lowercase = True						,   lowercase = None						,   lowercase = None						,   lowercase = None						,   lowercase = "max_length"						,   lowercase = None						,   lowercase = None						,   lowercase = None						,   **lowercase						,   ):

							if sampling_rate is not None:
											if sampling_rate != self.sampling_rate:
															raise ValueError(
															    F'''The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'''
															    F''' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'''
															    F''' was sampled with {self.sampling_rate} and not {sampling_rate}.'''						)
							else:
											logger.warning(
											    'It is strongly recommended to pass the `sampling_rate` argument to this function. '
											    'Failing to do so can result in silent errors that might be hard to debug.'						)

							_lowerCamelCase  :    List[Any]					=					isinstance(lowercase						,   np.ndarray						) and len(raw_speech.shape						) > 1
							if is_batched_numpy and len(raw_speech.shape						) > 2:
											raise ValueError(F'''Only mono-channel audio is supported for input to {self}'''						)
							_lowerCamelCase  :    List[Any]					=					is_batched_numpy or (
							    isinstance(lowercase						,   (list, tuple)						) and (isinstance(raw_speech[0]						,   (np.ndarray, tuple, list)						))
							)

							if is_batched:
											_lowerCamelCase  :    Any					=					[np.asarray([speech]						,   dtype=np.floataa						).T for speech in raw_speech]
							elif not is_batched and not isinstance(lowercase						,   np.ndarray						):
											_lowerCamelCase  :    Dict					=					np.asarray(lowercase						,   dtype=np.floataa						)
							elif isinstance(lowercase						,   np.ndarray						) and raw_speech.dtype is np.dtype(np.floataa						):
											_lowerCamelCase  :    int					=					raw_speech.astype(np.floataa						)

							# always return batch
							if not is_batched:
											_lowerCamelCase  :    List[Any]					=					[np.asarray([raw_speech]						).T]

							_lowerCamelCase  :    Optional[Any]					=					BatchFeature({'input_features': raw_speech}						)

							# convert into correct format for padding

							_lowerCamelCase  :    Optional[int]					=					self.pad(
							    lowercase						,   padding=lowercase						,   max_length=max_length if max_length else self.n_samples						,   truncation=lowercase						,   pad_to_multiple_of=lowercase						,   return_attention_mask=return_attention_mask or do_normalize						,   )

							# zero-mean and unit-variance normalization
							if do_normalize:
											_lowerCamelCase  :    Optional[int]					=					self.zero_mean_unit_var_norm(
											    padded_inputs['input_features']						,   attention_mask=padded_inputs['attention_mask']						,   padding_value=self.padding_value						,   )
											_lowerCamelCase  :    List[str]					=					np.stack(padded_inputs['input_features']						,   axis=0						)

							# make sure list is in array format
							_lowerCamelCase  :    Optional[Any]					=					padded_inputs.get('input_features'						).transpose(2						,   0						,   1						)

							_lowerCamelCase  :    Optional[int]					=					[self._np_extract_fbank_features(lowercase						) for waveform in input_features[0]]

							if isinstance(input_features[0]						,   lowercase						):
											_lowerCamelCase  :    Tuple					=					[np.asarray(lowercase						,   dtype=np.floataa						) for feature in input_features]
							else:
											_lowerCamelCase  :    str					=					input_features

							if return_attention_mask:
											# rescale from sample (48000) to feature (3000)
											_lowerCamelCase  :    Any					=					padded_inputs['attention_mask'][:, :: self.hop_length]

							if return_tensors is not None:
											_lowerCamelCase  :    Tuple					=					padded_inputs.convert_to_tensors(lowercase						)

							return padded_inputs






			def       A_					(						self						):
							_lowerCamelCase  :    Optional[Any]					=					copy.deepcopy(self.__dict__						)
							_lowerCamelCase  :    Optional[Any]					=					self.__class__.__name__
							if "mel_filters" in output:
											del output["mel_filters"]
							return output





"""simple docstring"""






import comet  # From: unbabel-comet
import torch

import datasets


lowercase__  		=		datasets.logging.get_logger(__name__)

lowercase__  		=		"""\
@inproceedings{rei-EtAl:2020:WMT,
   author    = {Rei, Ricardo  and  Stewart, Craig  and  Farinha, Ana C  and  Lavie, Alon},
   title     = {Unbabel's Participation in the WMT20 Metrics Shared Task},
   booktitle      = {Proceedings of the Fifth Conference on Machine Translation},
   month          = {November},
   year           = {2020},
   address        = {Online},
   publisher      = {Association for Computational Linguistics},
   pages     = {909--918},
}
@inproceedings{rei-etal-2020-comet,
   title = \"{COMET}: A Neural Framework for {MT} Evaluation\",
   author = \"Rei, Ricardo  and
      Stewart, Craig  and
      Farinha, Ana C  and
      Lavie, Alon\",
   booktitle = \"Proceedings of the 2020 Conference on Empirical Methods in Natural Language Processing (EMNLP)\",
   month = nov,
   year = \"2020\",
   address = \"Online\",
   publisher = \"Association for Computational Linguistics\",
   url = \"https://www.aclweb.org/anthology/2020.emnlp-main.213\",
   pages = \"2685--2702\",
}
"""

lowercase__  		=		"""\
Crosslingual Optimized Metric for Evaluation of Translation (COMET) is an open-source framework used to train Machine Translation metrics that achieve high levels of correlation with different types of human judgments (HTER, DA's or MQM).
With the release of the framework the authors also released fully trained models that were used to compete in the WMT20 Metrics Shared Task achieving SOTA in that years competition.

See the [README.md] file at https://unbabel.github.io/COMET/html/models.html for more information.
"""

lowercase__  		=		"""
COMET score.

Args:

`sources` (list of str): Source sentences
`predictions` (list of str): candidate translations
`references` (list of str): reference translations
`cuda` (bool): If set to True, runs COMET using GPU
`show_progress` (bool): Shows progress
`model`: COMET model to be used. Will default to `wmt-large-da-estimator-1719` if None.

Returns:
    `samples`: List of dictionaries with `src`, `mt`, `ref` and `score`.
    `scores`: List of scores.

Examples:

    >>> comet_metric = datasets.load_metric('comet')
    >>> # comet_metric = load_metric('comet', 'wmt20-comet-da')  # you can also choose which model to use
    >>> source = [\"Dem Feuer konnte Einhalt geboten werden\", \"Schulen und Kindergärten wurden eröffnet.\"]
    >>> hypothesis = [\"The fire could be stopped\", \"Schools and kindergartens were open\"]
    >>> reference = [\"They were able to control the fire.\", \"Schools and kindergartens opened\"]
    >>> results = comet_metric.compute(predictions=hypothesis, references=reference, sources=source)
    >>> print([round(v, 2) for v in results[\"scores\"]])
    [0.19, 0.92]
"""






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






			'''simple docstring'''



			def       A_					(						self						):
							return datasets.MetricInfo(
							    description=_DESCRIPTION						,   citation=_CITATION						,   homepage='https://unbabel.github.io/COMET/html/index.html'						,   inputs_description=_KWARGS_DESCRIPTION						,   features=datasets.Features(
							        {
							            'sources': datasets.Value('string'						,   id='sequence'						),
							            'predictions': datasets.Value('string'						,   id='sequence'						),
							            'references': datasets.Value('string'						,   id='sequence'						),
							        }						)						,   codebase_urls=['https://github.com/Unbabel/COMET']						,   reference_urls=[
							        'https://github.com/Unbabel/COMET',
							        'https://www.aclweb.org/anthology/2020.emnlp-main.213/',
							        'http://www.statmt.org/wmt20/pdf/2020.wmt-1.101.pdf6',
							    ]						,   )
			def       A_					(						self						,   lowercase						):
							if self.config_name == "default":
											_lowerCamelCase  :    List[Any]					=					comet.load_from_checkpoint(comet.download_model('wmt20-comet-da'						)						)
							else:
											_lowerCamelCase  :    Tuple					=					comet.load_from_checkpoint(comet.download_model(self.config_name						)						)






			def       A_					(						self						,   lowercase						,   lowercase						,   lowercase						,   lowercase=None						,   lowercase=False						):
							if gpus is None:
											_lowerCamelCase  :    Tuple					=					1 if torch.cuda.is_available() else 0
							_lowerCamelCase  :    List[Any]					=					{'src': sources, 'mt': predictions, 'ref': references}
							_lowerCamelCase  :    int					=					[dict(zip(lowercase						,   lowercase						)						) for t in zip(*data.values()						)]
							_lowerCamelCase,				_lowerCamelCase  :    List[str]					=					self.scorer.predict(lowercase						,   gpus=lowercase						,   progress_bar=lowercase						)
							return {"mean_score": mean_score, "scores": scores}