



"""simple docstring"""





from .hash_table import HashTable
from .number_theory.prime_numbers import is_prime, next_prime

class       __lowercase       (			lowerCAmelCase__	):


				'''simple docstring'''


				def __init__(   self   :     str      ,*SCREAMING_SNAKE_CASE_   :     List[Any]      ,**SCREAMING_SNAKE_CASE_   :     Dict      ):
										super().__init__(*_lowerCamelCase      ,**_lowerCamelCase      )

				def       _lowerCamelCase					(   self   :     int      ,SCREAMING_SNAKE_CASE_   :     Any      ,SCREAMING_SNAKE_CASE_   :     str      ):
										snake_case__       :					Union[str, Any]										=					(
										    next_prime(value % self.size_table      )
										    if not is_prime(value % self.size_table      )
										    else value % self.size_table
										)  # gt = bigger than
										return next_prime_gt - (data % next_prime_gt)

				def       _lowerCamelCase					(   self   :     Optional[int]      ,SCREAMING_SNAKE_CASE_   :     Optional[Any]      ,SCREAMING_SNAKE_CASE_   :     str      ,SCREAMING_SNAKE_CASE_   :     Any      ):
										return (increment * self.__hash_function_a(_lowerCamelCase      ,_lowerCamelCase      )) % self.size_table





				def       _lowerCamelCase					(   self   :     List[Any]      ,SCREAMING_SNAKE_CASE_   :     Dict      ,SCREAMING_SNAKE_CASE_   :     List[Any]=None      ):
										snake_case__       :					List[str]										=					1
										snake_case__       :					List[str]										=					self.hash_function(_lowerCamelCase      )

										while self.values[new_key] is not None and self.values[new_key] != key:
																snake_case__       :					Dict										=					(
																    self.__hash_double_function(_lowerCamelCase      ,_lowerCamelCase      ,_lowerCamelCase      )
																    if self.balanced_factor() >= self.lim_charge
																    else None
																)
																if new_key is None:
																						break
																else:
																						i += 1

										return new_key








"""simple docstring"""





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 TensorType, logging


_lowercase	      =    logging.get_logger(__name__)

class       __lowercase       (			snake_case_	):


				'''simple docstring'''


				_lowerCAmelCase				=["input_values", "attention_mask"]

				def __init__(   self   :     Tuple      ,SCREAMING_SNAKE_CASE_   :     Dict=1      ,SCREAMING_SNAKE_CASE_   :     List[str]=1_6_0_0_0      ,SCREAMING_SNAKE_CASE_   :     Any=1_2_8      ,SCREAMING_SNAKE_CASE_   :     List[str]=1_0_2_4      ,SCREAMING_SNAKE_CASE_   :     Any=0.0      ,SCREAMING_SNAKE_CASE_   :     Optional[Any]=True      ,SCREAMING_SNAKE_CASE_   :     Tuple=-4.2677393      ,SCREAMING_SNAKE_CASE_   :     List[Any]=4.5689974      ,SCREAMING_SNAKE_CASE_   :     str=False      ,**SCREAMING_SNAKE_CASE_   :     Dict      ,):
										super().__init__(feature_size=SCREAMING_SNAKE_CASE_      ,sampling_rate=SCREAMING_SNAKE_CASE_      ,padding_value=SCREAMING_SNAKE_CASE_      ,**SCREAMING_SNAKE_CASE_      )
										snake_case__       :					str										=					num_mel_bins
										snake_case__       :					Union[str, Any]										=					max_length
										snake_case__       :					List[str]										=					do_normalize
										snake_case__       :					Optional[int]										=					mean
										snake_case__       :					int										=					std
										snake_case__       :					Any										=					return_attention_mask

				def       _lowerCamelCase					(   self   :     Any      ,SCREAMING_SNAKE_CASE_   :     np.ndarray      ,SCREAMING_SNAKE_CASE_   :     int      ,):
										snake_case__       :					Any										=					torch.from_numpy(SCREAMING_SNAKE_CASE_      ).unsqueeze(0      )
										snake_case__       :					Optional[int]										=					ta_kaldi.fbank(
										    SCREAMING_SNAKE_CASE_      ,htk_compat=SCREAMING_SNAKE_CASE_      ,sample_frequency=self.sampling_rate      ,use_energy=SCREAMING_SNAKE_CASE_      ,window_type='''hanning'''      ,num_mel_bins=self.num_mel_bins      ,dither=0.0      ,frame_shift=1_0      ,)

										snake_case__       :					Dict										=					fbank.shape[0]
										snake_case__       :					Optional[int]										=					max_length - n_frames

										# pad or truncate, depending on difference
										if difference > 0:
																snake_case__       :					List[str]										=					torch.nn.ZeroPadad((0, 0, 0, difference)      )
																snake_case__       :					List[Any]										=					pad_module(SCREAMING_SNAKE_CASE_      )
										elif difference < 0:
																snake_case__       :					List[str]										=					fbank[0:max_length, :]

										snake_case__       :					Dict										=					fbank.numpy()

										return fbank

				def       _lowerCamelCase					(   self   :     Optional[int]      ,SCREAMING_SNAKE_CASE_   :     np.ndarray      ):
										return (input_values - (self.mean)) / (self.std * 2)





				def __call__(   self   :     Tuple      ,SCREAMING_SNAKE_CASE_   :     Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]      ,SCREAMING_SNAKE_CASE_   :     Optional[int] = None      ,SCREAMING_SNAKE_CASE_   :     Optional[Union[str, TensorType]] = None      ,**SCREAMING_SNAKE_CASE_   :     int      ,):

										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.'''      )

										snake_case__       :					Optional[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}'''      )
										snake_case__       :					Union[str, Any]										=					is_batched_numpy or (
										    isinstance(SCREAMING_SNAKE_CASE_      ,(list, tuple)      ) and (isinstance(raw_speech[0]      ,(np.ndarray, tuple, list)      ))
										)

										if is_batched:
																snake_case__       :					Optional[Any]										=					[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      ):
																snake_case__       :					Union[str, Any]										=					np.asarray(SCREAMING_SNAKE_CASE_      ,dtype=np.floataa      )
										elif isinstance(SCREAMING_SNAKE_CASE_      ,np.ndarray      ) and raw_speech.dtype is np.dtype(np.floataa      ):
																snake_case__       :					List[str]										=					raw_speech.astype(np.floataa      )

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

										# extract fbank features and pad/truncate to max_length
										snake_case__       :					List[Any]										=					[self._extract_fbank_features(SCREAMING_SNAKE_CASE_      ,max_length=self.max_length      ) for waveform in raw_speech]

										# convert into BatchFeature
										snake_case__       :					str										=					BatchFeature({'''input_values''': features}      )

										# make sure list is in array format
										snake_case__       :					Union[str, Any]										=					padded_inputs.get('''input_values'''      )
										if isinstance(input_values[0]      ,SCREAMING_SNAKE_CASE_      ):
																snake_case__       :					List[str]										=					[np.asarray(SCREAMING_SNAKE_CASE_      ,dtype=np.floataa      ) for feature in input_values]

										# normalization
										if self.do_normalize:
																snake_case__       :					List[Any]										=					[self.normalize(SCREAMING_SNAKE_CASE_      ) for feature in input_values]

										if return_tensors is not None:
																snake_case__       :					Union[str, Any]										=					padded_inputs.convert_to_tensors(SCREAMING_SNAKE_CASE_      )

										return padded_inputs


