



from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


__snake_case						:str      		=     logging.get_logger(__name__)

class     _A       (	__UpperCAmelCase    ):
       UpperCamelCase__  :    Optional[Any]											=						['''input_features''', '''attention_mask''']



       def __init__(		self							:     Any     ,		__SCREAMING_SNAKE_CASE							:     Optional[int]=80     ,		__SCREAMING_SNAKE_CASE							:     List[str]=16_000     ,		__SCREAMING_SNAKE_CASE							:     Dict=80     ,		__SCREAMING_SNAKE_CASE							:     str=0.0     ,		__SCREAMING_SNAKE_CASE							:     str=True     ,		__SCREAMING_SNAKE_CASE							:     str=True     ,		__SCREAMING_SNAKE_CASE							:     Tuple=True     ,		**__SCREAMING_SNAKE_CASE							:     Any     ,		):


           '''simple docstring'''

           super().__init__(feature_size=__SCREAMING_SNAKE_CASE     ,		sampling_rate=__SCREAMING_SNAKE_CASE     ,		padding_value=__SCREAMING_SNAKE_CASE     ,		**__SCREAMING_SNAKE_CASE)
           __a						       =				num_mel_bins
           __a						       =				do_ceptral_normalize
           __a						       =				normalize_means
           __a						       =				normalize_vars
           __a						       =				True



       def 		_lowerCamelCase			(		self							:     Optional[int]     ,		__SCREAMING_SNAKE_CASE							:     np.ndarray     ,		):


           '''simple docstring'''

           __a						       =				waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
           __a						       =				torch.from_numpy(__SCREAMING_SNAKE_CASE).unsqueeze(0)
           __a						       =				ta_kaldi.fbank(__SCREAMING_SNAKE_CASE     ,		num_mel_bins=self.num_mel_bins     ,		sample_frequency=self.sampling_rate)
           return features.numpy()



       @staticmethod
       def 		_lowerCamelCase			(		__SCREAMING_SNAKE_CASE							:     np.ndarray     ,		__SCREAMING_SNAKE_CASE							:     int     ,		__SCREAMING_SNAKE_CASE							:     Optional[bool] = True     ,		__SCREAMING_SNAKE_CASE							:     Optional[bool] = True     ,		__SCREAMING_SNAKE_CASE							:     float = 0.0     ,		):


           '''simple docstring'''

           if normalize_means:
               __a						       =				x[:input_length].mean(axis=0)
               __a						       =				np.subtract(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE)
           if normalize_vars:
               __a						       =				x[:input_length].std(axis=0)
               __a						       =				np.divide(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE)

           if input_length < x.shape[0]:
               __a						       =				padding_value

           # make sure array is in float32
           __a						       =				x.astype(np.floataa)

           return x



       def 		_lowerCamelCase			(		self							:     int     ,		__SCREAMING_SNAKE_CASE							:     List[np.ndarray]     ,		__SCREAMING_SNAKE_CASE							:     Optional[np.ndarray] = None):


           '''simple docstring'''

           __a						       =				attention_mask.sum(-1) if attention_mask is not None else [x.shape[0] for x in input_features]
           return [
               self.utterance_cmvn(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		self.normalize_means     ,		self.normalize_vars     ,		self.padding_value)
               for x, n in zip(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE)
           ]





       def __call__(		self							:     Optional[Any]     ,		__SCREAMING_SNAKE_CASE							:     Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]     ,		__SCREAMING_SNAKE_CASE							:     Union[bool, str, PaddingStrategy] = False     ,		__SCREAMING_SNAKE_CASE							:     Optional[int] = None     ,		__SCREAMING_SNAKE_CASE							:     bool = False     ,		__SCREAMING_SNAKE_CASE							:     Optional[int] = None     ,		__SCREAMING_SNAKE_CASE							:     Optional[Union[str, TensorType]] = None     ,		__SCREAMING_SNAKE_CASE							:     Optional[int] = None     ,		__SCREAMING_SNAKE_CASE							:     Optional[bool] = None     ,		**__SCREAMING_SNAKE_CASE							:     Dict     ,		):


           '''simple docstring'''

           if sampling_rate is not None:
               if sampling_rate != self.sampling_rate:
                   raise ValueError(
                       F'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
                       F' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'
                       F' {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.''')

           __a						       =				isinstance(__SCREAMING_SNAKE_CASE     ,		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}')
           __a						       =				is_batched_numpy or (
               isinstance(__SCREAMING_SNAKE_CASE     ,		(list, tuple)) and (isinstance(raw_speech[0]     ,		(np.ndarray, tuple, list)))
           )

           if is_batched:
               __a						       =				[np.asarray(__SCREAMING_SNAKE_CASE     ,		dtype=np.floataa) for speech in raw_speech]
           elif not is_batched and not isinstance(__SCREAMING_SNAKE_CASE     ,		np.ndarray):
               __a						       =				np.asarray(__SCREAMING_SNAKE_CASE     ,		dtype=np.floataa)
           elif isinstance(__SCREAMING_SNAKE_CASE     ,		np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
               __a						       =				raw_speech.astype(np.floataa)

           # always return batch
           if not is_batched:
               __a						       =				[raw_speech]

           # extract fbank features
           __a						       =				[self._extract_fbank_features(__SCREAMING_SNAKE_CASE) for waveform in raw_speech]

           # convert into correct format for padding
           __a						       =				BatchFeature({'''input_features''': features})

           __a						       =				self.pad(
               __SCREAMING_SNAKE_CASE     ,		padding=__SCREAMING_SNAKE_CASE     ,		max_length=__SCREAMING_SNAKE_CASE     ,		truncation=__SCREAMING_SNAKE_CASE     ,		pad_to_multiple_of=__SCREAMING_SNAKE_CASE     ,		return_attention_mask=__SCREAMING_SNAKE_CASE     ,		**__SCREAMING_SNAKE_CASE     ,		)

           # make sure list is in array format
           __a						       =				padded_inputs.get('''input_features''')
           if isinstance(input_features[0]     ,		__SCREAMING_SNAKE_CASE):
               __a						       =				[np.asarray(__SCREAMING_SNAKE_CASE     ,		dtype=np.floataa) for feature in input_features]

           __a						       =				padded_inputs.get('''attention_mask''')
           if attention_mask is not None:
               __a						       =				[np.asarray(__SCREAMING_SNAKE_CASE     ,		dtype=np.intaa) for array in attention_mask]

           # Utterance-level cepstral mean and variance normalization
           if self.do_ceptral_normalize:
               __a						       =				(
                   np.array(__SCREAMING_SNAKE_CASE     ,		dtype=np.intaa)
                   if self._get_padding_strategies(__SCREAMING_SNAKE_CASE     ,		max_length=__SCREAMING_SNAKE_CASE) is not PaddingStrategy.DO_NOT_PAD
                   else None
               )
               __a						       =				self.normalize(
                   padded_inputs['''input_features''']     ,		attention_mask=__SCREAMING_SNAKE_CASE)

           if return_tensors is not None:
               __a						       =				padded_inputs.convert_to_tensors(__SCREAMING_SNAKE_CASE)

           return padded_inputs










from typing import Optional

from .. import Features, NamedSplit
from ..packaged_modules.text.text import Text
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader

class     _A       (	__UpperCAmelCase    ):



       def __init__(		self							:     Optional[int]     ,		__SCREAMING_SNAKE_CASE							:     NestedDataStructureLike[PathLike]     ,		__SCREAMING_SNAKE_CASE							:     Optional[NamedSplit] = None     ,		__SCREAMING_SNAKE_CASE							:     Optional[Features] = None     ,		__SCREAMING_SNAKE_CASE							:     str = None     ,		__SCREAMING_SNAKE_CASE							:     bool = False     ,		__SCREAMING_SNAKE_CASE							:     bool = False     ,		__SCREAMING_SNAKE_CASE							:     Optional[int] = None     ,		**__SCREAMING_SNAKE_CASE							:     List[str]     ,		):


           '''simple docstring'''

           super().__init__(
               __SCREAMING_SNAKE_CASE     ,		split=__SCREAMING_SNAKE_CASE     ,		features=__SCREAMING_SNAKE_CASE     ,		cache_dir=__SCREAMING_SNAKE_CASE     ,		keep_in_memory=__SCREAMING_SNAKE_CASE     ,		streaming=__SCREAMING_SNAKE_CASE     ,		num_proc=__SCREAMING_SNAKE_CASE     ,		**__SCREAMING_SNAKE_CASE     ,		)
           __a						       =				path_or_paths if isinstance(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE) else {self.split: path_or_paths}
           __a						       =				Text(
               cache_dir=__SCREAMING_SNAKE_CASE     ,		data_files=__SCREAMING_SNAKE_CASE     ,		features=__SCREAMING_SNAKE_CASE     ,		**__SCREAMING_SNAKE_CASE     ,		)





       def 		_lowerCamelCase			(		self							:     List[str]):


           '''simple docstring'''

           if self.streaming:
               __a						       =				self.builder.as_streaming_dataset(split=self.split)
           # Build regular (map-style) dataset
           else:
               __a						       =				None
               __a						       =				None
               __a						       =				None
               __a						       =				None

               self.builder.download_and_prepare(
                   download_config=__SCREAMING_SNAKE_CASE     ,		download_mode=__SCREAMING_SNAKE_CASE     ,		verification_mode=__SCREAMING_SNAKE_CASE     ,		base_path=__SCREAMING_SNAKE_CASE     ,		num_proc=self.num_proc     ,		)
               __a						       =				self.builder.as_dataset(
                   split=self.split     ,		verification_mode=__SCREAMING_SNAKE_CASE     ,		in_memory=self.keep_in_memory)
           return dataset





