"""simple docstring"""


import numpy as np

import datasets


lowerCAmelCase_:					Any     =					"\nCompute the Mahalanobis Distance\n\nMahalonobis distance is the distance between a point and a distribution.\nAnd not between two distinct points. It is effectively a multivariate equivalent of the Euclidean distance.\nIt was introduced by Prof. P. C. Mahalanobis in 1936\nand has been used in various statistical applications ever since\n[source: https://www.machinelearningplus.com/statistics/mahalanobis-distance/]\n"

lowerCAmelCase_:					Optional[int]     =					"\\n@article{de2000mahalanobis,\n  title={The mahalanobis distance},\n  author={De Maesschalck, Roy and Jouan-Rimbaud, Delphine and Massart, D{\'e}sir{\'e} L},\n  journal={Chemometrics and intelligent laboratory systems},\n  volume={50},\n  number={1},\n  pages={1--18},\n  year={2000},\n  publisher={Elsevier}\n}\n"

lowerCAmelCase_:					List[str]     =					"\nArgs:\n    X: List of datapoints to be compared with the `reference_distribution`.\n    reference_distribution: List of datapoints from the reference distribution we want to compare to.\nReturns:\n    mahalanobis: The Mahalonobis distance for each datapoint in `X`.\nExamples:\n\n    >>> mahalanobis_metric = datasets.load_metric(\"mahalanobis\")\n    >>> results = mahalanobis_metric.compute(reference_distribution=[[0, 1], [1, 0]], X=[[0, 1]])\n    >>> print(results)\n    {'mahalanobis': array([0.5])}\n"






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


   def 	snake_case__	(     self  ):
        '''simple docstring'''



        return datasets.MetricInfo(
            description=_DESCRIPTION,				citation=_CITATION,				inputs_description=_KWARGS_DESCRIPTION,				features=datasets.Features(
                {
                    "X": datasets.Sequence(datasets.Value("float",				id="sequence"  ),				id="X"  ),
                }  ),				)


   def 	snake_case__	(     self,				_UpperCAmelCase,				_UpperCAmelCase  ):
        '''simple docstring'''



        lowercase__				 =							np.array(_UpperCAmelCase  )
        lowercase__				 =							np.array(_UpperCAmelCase  )

        # Assert that arrays are 2D
        if len(X.shape  ) != 2:
             raise ValueError("Expected `X` to be a 2D vector"  )
        if len(reference_distribution.shape  ) != 2:
             raise ValueError("Expected `reference_distribution` to be a 2D vector"  )
        if reference_distribution.shape[0] < 2:
             raise ValueError(
                 "Expected `reference_distribution` to be a 2D vector with more than one element in the first dimension"  )

        # Get mahalanobis distance for each prediction
        lowercase__				 =							X - np.mean(_UpperCAmelCase  )
        lowercase__				 =							np.cov(reference_distribution.T  )
        try:
             lowercase__				 =							np.linalg.inv(_UpperCAmelCase  )
        except np.linalg.LinAlgError:
             lowercase__				 =							np.linalg.pinv(_UpperCAmelCase  )
        lowercase__				 =							np.dot(_UpperCAmelCase,				_UpperCAmelCase  )
        lowercase__				 =							np.dot(_UpperCAmelCase,				X_minus_mu.T  ).diagonal()

        return {"mahalanobis": mahal_dist}





"""simple docstring"""


from math import ceil
from typing import List, Optional, Union

import numpy as np

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


lowerCAmelCase_:					Union[str, Any]     =					logging.get_logger(__name__)






class    a__    (   _a	):
   snake_case_							=      ["audio_values", "audio_mask"]


   def __init__(     self,				_UpperCAmelCase=2048,				_UpperCAmelCase=1,				_UpperCAmelCase=[16, 16],				_UpperCAmelCase=128,				_UpperCAmelCase=4_4100,				_UpperCAmelCase=86,				_UpperCAmelCase=2048,				_UpperCAmelCase=0.0,				**_UpperCAmelCase,				):
        '''simple docstring'''



        super().__init__(
            feature_size=_UpperCAmelCase,				sampling_rate=_UpperCAmelCase,				padding_value=_UpperCAmelCase,				**_UpperCAmelCase,				)

        lowercase__				 =							spectrogram_length
        lowercase__				 =							num_channels
        lowercase__				 =							patch_size
        lowercase__				 =							feature_size // self.patch_size[1]
        lowercase__				 =							n_fft
        lowercase__				 =							sampling_rate // hop_length_to_sampling_rate
        lowercase__				 =							sampling_rate
        lowercase__				 =							padding_value
        lowercase__				 =							mel_filter_bank(
            num_frequency_bins=1 + n_fft // 2,				num_mel_filters=_UpperCAmelCase,				min_frequency=0.0,				max_frequency=22_050.0,				sampling_rate=_UpperCAmelCase,				norm="slaney",				mel_scale="slaney",				).T


   def 	snake_case__	(     self,				_UpperCAmelCase  ):
        '''simple docstring'''



        lowercase__				 =							spectrogram(
            _UpperCAmelCase,				window_function(self.n_fft,				"hann"  ),				frame_length=self.n_fft,				hop_length=self.hop_length,				power=2.0,				mel_filters=self.mel_filters.T,				log_mel="dB",				db_range=80.0,				)
        lowercase__				 =							log_spec[:, :-1]
        lowercase__				 =							log_spec - 20.0
        lowercase__				 =							np.clip(log_spec / 40.0,				-2.0,				0.0  ) + 1.0
        return log_spec


   def __call__(     self,				_UpperCAmelCase,				_UpperCAmelCase = None,				_UpperCAmelCase = True,				_UpperCAmelCase = None,				_UpperCAmelCase = False,				_UpperCAmelCase = False,				**_UpperCAmelCase,				):
        '''simple docstring'''



        if sampling_rate is not None:
             if sampling_rate != self.sampling_rate:
                  raise ValueError(
                      "This feature extractor is set to support sampling rate"
                      F''' of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled'''
                      F''' 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."  )

        lowercase__				 =							isinstance(_UpperCAmelCase,				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}'''  )
        lowercase__				 =							is_batched_numpy or (
            isinstance(_UpperCAmelCase,				(list, tuple)  ) and (isinstance(raw_speech[0],				(np.ndarray, tuple, list)  ))
        )
        if is_batched:
             lowercase__				 =							[np.asarray([speech],				dtype=np.floataa  ).T for speech in raw_speech]
        elif not is_batched and not isinstance(_UpperCAmelCase,				np.ndarray  ):
             lowercase__				 =							np.asarray(_UpperCAmelCase,				dtype=np.floataa  )
        elif isinstance(_UpperCAmelCase,				np.ndarray  ) and raw_speech.dtype is np.dtype(np.floataa  ):
             lowercase__				 =							raw_speech.astype(np.floataa  )
        # always return batch
        if not is_batched:
             lowercase__				 =							[np.asarray([raw_speech]  ).T]

        # Convert audio signals to log mel spectrograms, truncate by time axis
        lowercase__				 =							[
            self._np_extract_fbank_features(waveform.squeeze()  ).T[: self.spectrogram_length] for waveform in raw_speech
        ]
        if isinstance(audio_features[0],				_UpperCAmelCase  ):
             lowercase__				 =							[np.asarray(_UpperCAmelCase,				dtype=np.floataa  ) for feature in audio_features]

        # Create audio attention mask
        lowercase__				 =							max(
            [ceil(feature.shape[0] / self.patch_size[0]  ) * self.freq_len for feature in audio_features]  )  # The maximum number of audio patches in a batch
        if return_attention_mask:
             lowercase__				 =							[
                 (ceil(feature.shape[0] / self.patch_size[0]  ) * self.freq_len) * [1]
                 + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]  ) * self.freq_len) * [0]
                 for feature in audio_features
             ]
             lowercase__				 =							np.array(_UpperCAmelCase  ).astype(np.floataa  )

        # convert into correct format for padding
        lowercase__				 =							max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
        lowercase__				 =							np.ones([len(_UpperCAmelCase  ), 1, max_time_len, self.feature_size]  ).astype(np.floataa  )
        lowercase__				 =							padded_audio_features * self.padding_value
        for i in range(len(_UpperCAmelCase  )  ):
             lowercase__				 =							audio_features[i]
             lowercase__				 =							feature

        # return as BatchFeature
        if return_attention_mask:
             lowercase__				 =							{"audio_values": padded_audio_features, "audio_mask": audio_mask}
        else:
             lowercase__				 =							{"audio_values": padded_audio_features}

        lowercase__				 =							BatchFeature(data=_UpperCAmelCase,				tensor_type=_UpperCAmelCase  )
        return encoded_inputs




