


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


__A	=logging.get_logger(__name__)


class 		_snake_case					(     __UpperCAmelCase	):
					lowerCAmelCase      :Optional[Any]   =					['''input_features''', '''attention_mask''']




					def __init__(      self							,							_lowerCamelCase=80							,							_lowerCamelCase=1_6000							,							_lowerCamelCase=80							,							_lowerCamelCase=0.0							,							_lowerCamelCase=True							,							_lowerCamelCase=True							,							_lowerCamelCase=True							,							**_lowerCamelCase							,							):
								super().__init__(feature_size=lowerCAmelCase_							,							sampling_rate=lowerCAmelCase_							,							padding_value=lowerCAmelCase_							,							**lowerCAmelCase_)
								UpperCAmelCase__  :					List[str]        =     num_mel_bins
								UpperCAmelCase__  :					str        =     do_ceptral_normalize
								UpperCAmelCase__  :					Tuple        =     normalize_means
								UpperCAmelCase__  :					Dict        =     normalize_vars
								UpperCAmelCase__  :					Dict        =     True




					def 				snake_case__						(      self							,							_lowerCamelCase							,							):
								UpperCAmelCase__  :					str        =     waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
								UpperCAmelCase__  :					Any        =     torch.from_numpy(lowerCAmelCase_).unsqueeze(0)
								UpperCAmelCase__  :					Optional[Any]        =     ta_kaldi.fbank(lowerCAmelCase_							,							num_mel_bins=self.num_mel_bins							,							sample_frequency=self.sampling_rate)
								return features.numpy()




					@staticmethod
					def 				snake_case__						(      _lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase = True							,							_lowerCamelCase = True							,							_lowerCamelCase = 0.0							,							):







								if normalize_means:
											UpperCAmelCase__  :					int        =     x[:input_length].mean(axis=0)
											UpperCAmelCase__  :					Optional[Any]        =     np.subtract(lowerCAmelCase_							,							lowerCAmelCase_)
								if normalize_vars:
											UpperCAmelCase__  :					List[Any]        =     x[:input_length].std(axis=0)
											UpperCAmelCase__  :					int        =     np.divide(lowerCAmelCase_							,							lowerCAmelCase_)

								if input_length < x.shape[0]:
											UpperCAmelCase__  :					List[str]        =     padding_value

								# make sure array is in float32
								UpperCAmelCase__  :					List[str]        =     x.astype(np.floataa)

								return x




					def 				snake_case__						(      self							,							_lowerCamelCase							,							_lowerCamelCase = None):
								UpperCAmelCase__  :					Tuple        =     attention_mask.sum(-1) if attention_mask is not None else [x.shape[0] for x in input_features]
								return [
								    self.utterance_cmvn(lowerCAmelCase_							,							lowerCAmelCase_							,							self.normalize_means							,							self.normalize_vars							,							self.padding_value)
								    for x, n in zip(lowerCAmelCase_							,							lowerCAmelCase_)
								]




					def __call__(      self							,							_lowerCamelCase							,							_lowerCamelCase = False							,							_lowerCamelCase = None							,							_lowerCamelCase = False							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							**_lowerCamelCase							,							):







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

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

								if is_batched:
											UpperCAmelCase__  :					Dict        =     [np.asarray(lowerCAmelCase_							,							dtype=np.floataa) for speech in raw_speech]
								elif not is_batched and not isinstance(lowerCAmelCase_							,							np.ndarray):
											UpperCAmelCase__  :					Dict        =     np.asarray(lowerCAmelCase_							,							dtype=np.floataa)
								elif isinstance(lowerCAmelCase_							,							np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
											UpperCAmelCase__  :					List[Any]        =     raw_speech.astype(np.floataa)

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

								# extract fbank features
								UpperCAmelCase__  :					str        =     [self._extract_fbank_features(lowerCAmelCase_) for waveform in raw_speech]

								# convert into correct format for padding
								UpperCAmelCase__  :					Dict        =     BatchFeature({"""input_features""": features})

								UpperCAmelCase__  :					int        =     self.pad(
								    lowerCAmelCase_							,							padding=lowerCAmelCase_							,							max_length=lowerCAmelCase_							,							truncation=lowerCAmelCase_							,							pad_to_multiple_of=lowerCAmelCase_							,							return_attention_mask=lowerCAmelCase_							,							**lowerCAmelCase_							,							)

								# make sure list is in array format
								UpperCAmelCase__  :					Optional[Any]        =     padded_inputs.get("""input_features""")
								if isinstance(input_features[0]							,							lowerCAmelCase_):
											UpperCAmelCase__  :					Dict        =     [np.asarray(lowerCAmelCase_							,							dtype=np.floataa) for feature in input_features]

								UpperCAmelCase__  :					Any        =     padded_inputs.get("""attention_mask""")
								if attention_mask is not None:
											UpperCAmelCase__  :					Dict        =     [np.asarray(lowerCAmelCase_							,							dtype=np.intaa) for array in attention_mask]

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

								if return_tensors is not None:
											UpperCAmelCase__  :					Any        =     padded_inputs.convert_to_tensors(lowerCAmelCase_)

								return padded_inputs



'''simple docstring'''



import re

from flax.core.frozen_dict import freeze
from flax.traverse_util import flatten_dict, unflatten_dict
from jax.experimental import PartitionSpec as P


# Sentinels
__A	=object()

# For specifying empty leaf dict `{}`
__A	=object()







def     _UpperCamelCase      (		UpperCamelCase__ ,      UpperCamelCase__					):
			UpperCAmelCase__  :					Union[str, Any]        =     tuple((re.compile(x + """$"""					) for x in qs)					)
			for i in range(len(UpperCamelCase__					) - len(UpperCamelCase__					) + 1					):
						UpperCAmelCase__  :					Optional[Any]        =     [x.match(UpperCamelCase__					) for x, y in zip(UpperCamelCase__ ,      ks[i:]					)]
						if matches and all(UpperCamelCase__					):
									return True
			return False







def     _UpperCamelCase      (		UpperCamelCase__					):
			def replace(UpperCamelCase__ ,      UpperCamelCase__					):
						for rule, replacement in rules:
									if _match(UpperCamelCase__ ,      UpperCamelCase__					):
												return replacement
						return val

			return replace







def     _UpperCamelCase      (		):
			return [
			    # embeddings
			    (("transformer", "wpe", "embedding"), P("""mp""" ,      UpperCamelCase__					)),
			    (("transformer", "wte", "embedding"), P("""mp""" ,      UpperCamelCase__					)),
			    # atention
			    (("attention", "(q_proj|k_proj|v_proj)", "kernel"), P(UpperCamelCase__ ,      """mp"""					)),
			    (("attention", "out_proj", "kernel"), P("""mp""" ,      UpperCamelCase__					)),
			    (("attention", "out_proj", "bias"), None),
			    # mlp
			    (("mlp", "c_fc", "kernel"), P(UpperCamelCase__ ,      """mp"""					)),
			    (("mlp", "c_fc", "bias"), P("""mp"""					)),
			    (("mlp", "c_proj", "kernel"), P("""mp""" ,      UpperCamelCase__					)),
			    (("mlp", "c_proj", "bias"), None),
			    # layer norms
			    ((r"ln_\d+", "bias"), None),
			    ((r"\d+", r"ln_\d+", "scale"), None),
			    (("ln_f", "bias"), None),
			    (("ln_f", "scale"), None),
			]







def     _UpperCamelCase      (		UpperCamelCase__					):
			UpperCAmelCase__  :					Optional[int]        =     _get_partition_rules()
			UpperCAmelCase__  :					Optional[int]        =     _replacement_rules(UpperCamelCase__					)
			UpperCAmelCase__  :					List[Any]        =     {k: _unmatched for k in flatten_dict(UpperCamelCase__					)}
			UpperCAmelCase__  :					Tuple        =     {k: replace(UpperCamelCase__ ,      UpperCamelCase__					) for k, v in initd.items()}
			assert _unmatched not in result.values(), "Incomplete partition spec."
			return freeze(unflatten_dict(UpperCamelCase__					)					)