


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


UpperCAmelCase_      : Optional[int]  					=       logging.get_logger(__name__)
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      snake_case__				:     str        =						['''input_features''', '''attention_mask''']
      def __init__(       self				:       Any     ,							SCREAMING_SNAKE_CASE__				:       Tuple=8_0     ,							SCREAMING_SNAKE_CASE__				:       Dict=1_6_0_0_0     ,							SCREAMING_SNAKE_CASE__				:       List[Any]=8_0     ,							SCREAMING_SNAKE_CASE__				:       Union[str, Any]=0.0     ,							SCREAMING_SNAKE_CASE__				:       Dict=True     ,							SCREAMING_SNAKE_CASE__				:       List[str]=True     ,							SCREAMING_SNAKE_CASE__				:       int=True     ,							**SCREAMING_SNAKE_CASE__				:       Optional[int]     ,							)		->			List[Any]:
             super().__init__(feature_size=SCREAMING_SNAKE_CASE__     ,							sampling_rate=SCREAMING_SNAKE_CASE__     ,							padding_value=SCREAMING_SNAKE_CASE__     ,							**SCREAMING_SNAKE_CASE__     )
             a_					:		List[Any]							=  num_mel_bins
             a_					:		Any							=  do_ceptral_normalize
             a_					:		Optional[Any]							=  normalize_means
             a_					:		Dict							=  normalize_vars
             a_					:		Union[str, Any]							=  True
      def        SCREAMING_SNAKE_CASE							(       self				:       Any     ,							SCREAMING_SNAKE_CASE__				:       np.ndarray     ,							)		->			np.ndarray:
             a_					:		Optional[int]							=  waveform * (2**1_5)  # Kaldi compliance: 16-bit signed integers
             a_					:		Union[str, Any]							=  torch.from_numpy(SCREAMING_SNAKE_CASE__     ).unsqueeze(0     )
             a_					:		Optional[Any]							=  ta_kaldi.fbank(SCREAMING_SNAKE_CASE__     ,							num_mel_bins=self.num_mel_bins     ,							sample_frequency=self.sampling_rate     )
             return features.numpy()
      @staticmethod
      def        SCREAMING_SNAKE_CASE							(       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     ,							)		->			np.ndarray:
             # make sure we normalize float32 arrays
             if normalize_means:
                    a_					:		Union[str, Any]							=  x[:input_length].mean(axis=0     )
                    a_					:		List[Any]							=  np.subtract(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     )
             if normalize_vars:
                    a_					:		Tuple							=  x[:input_length].std(axis=0     )
                    a_					:		str							=  np.divide(SCREAMING_SNAKE_CASE__     ,							SCREAMING_SNAKE_CASE__     )

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

             # make sure array is in float32
             a_					:		str							=  x.astype(np.floataa     )

             return x
      def        SCREAMING_SNAKE_CASE							(       self				:       List[str]     ,							SCREAMING_SNAKE_CASE__				:       List[np.ndarray]     ,							SCREAMING_SNAKE_CASE__				:       Optional[np.ndarray] = None     )		->			List[np.ndarray]:
             a_					:		Union[str, Any]							=  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				:       Union[str, 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__				:       Any     ,							)		->			BatchFeature:

             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_					:		Any							=  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_					:		Any							=  is_batched_numpy or (
                 isinstance(SCREAMING_SNAKE_CASE__     ,							(list, tuple)     ) and (isinstance(raw_speech[0]     ,							(np.ndarray, tuple, list)     ))
             )

             if is_batched:
                    a_					:		Optional[int]							=  [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_					:		Tuple							=  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_					:		Optional[int]							=  raw_speech.astype(np.floataa     )

             # always return batch
             if not is_batched:
                    a_					:		List[str]							=  [raw_speech]

             # extract fbank features
             a_					:		Any							=  [self._extract_fbank_features(SCREAMING_SNAKE_CASE__     ) for waveform in raw_speech]

             # convert into correct format for padding
             a_					:		List[Any]							=  BatchFeature({'input_features': features}     )

             a_					:		List[Any]							=  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_					:		Any							=  padded_inputs.get('input_features'     )
             if isinstance(input_features[0]     ,							SCREAMING_SNAKE_CASE__     ):
                    a_					:		Union[str, Any]							=  [np.asarray(SCREAMING_SNAKE_CASE__     ,							dtype=np.floataa     ) for feature in input_features]

             a_					:		str							=  padded_inputs.get('attention_mask'     )
             if attention_mask is not None:
                    a_					:		List[Any]							=  [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_					:		List[str]							=  (
                        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_					:		List[str]							=  self.normalize(
                        padded_inputs['input_features']     ,							attention_mask=SCREAMING_SNAKE_CASE__     )

             if return_tensors is not None:
                    a_					:		List[str]							=  padded_inputs.convert_to_tensors(SCREAMING_SNAKE_CASE__     )

             return padded_inputs







from collections import deque

from .hash_table import HashTable




class 					snake_case__						(       lowerCAmelCase_     ):






     """simple docstring"""


     def __init__(       self  :  Optional[Any], *_snake_case  :  Optional[Any], **_snake_case  :  List[Any]					)   ->Optional[int]:
            super().__init__(*_snake_case, **_snake_case					)






     def    lowercase_   (       self  :  Optional[Any], _snake_case  :  Tuple, _snake_case  :  Dict					)   ->Dict:
            snake_case__							:	int   	=	deque([]					) if self.values[key] is None else self.values[key]
            self.values[key].appendleft(_snake_case					)
            snake_case__							:	Dict   	=	self.values[key]






     def    lowercase_   (       self  :  Any					)   ->Optional[Any]:
            return (
                sum(self.charge_factor - len(_snake_case					) for slot in self.values					)
                / self.size_table
                * self.charge_factor
            )






     def    lowercase_   (       self  :  Union[str, Any], _snake_case  :  str, _snake_case  :  Optional[int]=None					)   ->Optional[Any]:
            if not (
                len(self.values[key]					) == self.charge_factor and self.values.count(_snake_case					) == 0
            ):
                   return key
            return super()._collision_resolution(_snake_case, _snake_case					)






