


"""simple docstring"""







import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

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


a__    :							List[str]            =					logging.get_logger(__name__)




class 			UpperCamelCase_     (			a__):

	"""simple docstring"""





	snake_case__    :  Optional[Any]    =      ["input_features"]




	def __init__(			self      :    Tuple      ,      UpperCAmelCase__      :    List[Any]=8_0      ,      UpperCAmelCase__      :    List[str]=1_6_0_0_0      ,      UpperCAmelCase__      :    List[Any]=1_6_0      ,      UpperCAmelCase__      :    Union[str, Any]=3_0      ,      UpperCAmelCase__      :    Optional[Any]=4_0_0      ,      UpperCAmelCase__      :    Union[str, Any]=0.0      ,      UpperCAmelCase__      :    str=False      ,      **UpperCAmelCase__      :    Optional[Any]      ,      )	->     Optional[Any]:
						super().__init__(
						    feature_size=UpperCAmelCase__      ,      sampling_rate=UpperCAmelCase__      ,      padding_value=UpperCAmelCase__      ,      return_attention_mask=UpperCAmelCase__      ,      **UpperCAmelCase__      ,      )
						__SCREAMING_SNAKE_CASE						       =    n_fft
						__SCREAMING_SNAKE_CASE						       =    hop_length
						__SCREAMING_SNAKE_CASE						       =    chunk_length
						__SCREAMING_SNAKE_CASE						       =    chunk_length * sampling_rate
						__SCREAMING_SNAKE_CASE						       =    self.n_samples // hop_length
						__SCREAMING_SNAKE_CASE						       =    sampling_rate
						__SCREAMING_SNAKE_CASE						       =    mel_filter_bank(
						    num_frequency_bins=1 + n_fft // 2      ,      num_mel_filters=UpperCAmelCase__      ,      min_frequency=0.0      ,      max_frequency=8_0_0_0.0      ,      sampling_rate=UpperCAmelCase__      ,      norm="slaney"      ,      mel_scale="slaney"      ,      )




	def   UpperCAmelCase_				(			self      :    List[Any]      ,      UpperCAmelCase__      :    np.array       )	->     List[str]:
						__SCREAMING_SNAKE_CASE						       =    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      ,      log_mel="log10"      ,      )
						__SCREAMING_SNAKE_CASE						       =    log_spec[:, :-1]
						__SCREAMING_SNAKE_CASE						       =    np.maximum(UpperCAmelCase__      ,      log_spec.max() - 8.0       )
						__SCREAMING_SNAKE_CASE						       =    (log_spec + 4.0) / 4.0
						return log_spec




	@staticmethod
	# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
	def   UpperCAmelCase_				(			UpperCAmelCase__      :    List[np.ndarray]      ,      UpperCAmelCase__      :    List[np.ndarray]      ,      UpperCAmelCase__      :    float = 0.0       )	->     Tuple:
						if attention_mask is not None:
											__SCREAMING_SNAKE_CASE						       =    np.array(UpperCAmelCase__      ,      np.intaa       )
											__SCREAMING_SNAKE_CASE						       =    []

											for vector, length in zip(UpperCAmelCase__      ,      attention_mask.sum(-1       )       ):
																__SCREAMING_SNAKE_CASE						       =    (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7       )
																if length < normed_slice.shape[0]:
																					__SCREAMING_SNAKE_CASE						       =    padding_value

																normed_input_values.append(UpperCAmelCase__       )
						else:
											__SCREAMING_SNAKE_CASE						       =    [(x - x.mean()) / np.sqrt(x.var() + 1E-7       ) for x in input_values]

						return normed_input_values




	def __call__(			self      :    Optional[int]      ,      UpperCAmelCase__      :    Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]      ,      UpperCAmelCase__      :    bool = True      ,      UpperCAmelCase__      :    Optional[int] = None      ,      UpperCAmelCase__      :    Optional[Union[str, TensorType]] = None      ,      UpperCAmelCase__      :    Optional[bool] = None      ,      UpperCAmelCase__      :    Optional[str] = "max_length"      ,      UpperCAmelCase__      :    Optional[int] = None      ,      UpperCAmelCase__      :    Optional[int] = None      ,      UpperCAmelCase__      :    Optional[bool] = None      ,      **UpperCAmelCase__      :    Tuple      ,      )	->     List[Any]:

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

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

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

						# always return batch
						if not is_batched:
											__SCREAMING_SNAKE_CASE						       =    [np.asarray([raw_speech]       ).T]

						__SCREAMING_SNAKE_CASE						       =    BatchFeature({"input_features": raw_speech}       )

						# convert into correct format for padding

						__SCREAMING_SNAKE_CASE						       =    self.pad(
						    UpperCAmelCase__      ,      padding=UpperCAmelCase__      ,      max_length=max_length if max_length else self.n_samples      ,      truncation=UpperCAmelCase__      ,      pad_to_multiple_of=UpperCAmelCase__      ,      return_attention_mask=return_attention_mask or do_normalize      ,      )

						# zero-mean and unit-variance normalization
						if do_normalize:
											__SCREAMING_SNAKE_CASE						       =    self.zero_mean_unit_var_norm(
											    padded_inputs["input_features"]      ,      attention_mask=padded_inputs["attention_mask"]      ,      padding_value=self.padding_value      ,      )
											__SCREAMING_SNAKE_CASE						       =    np.stack(padded_inputs["input_features"]      ,      axis=0       )

						# make sure list is in array format
						__SCREAMING_SNAKE_CASE						       =    padded_inputs.get("input_features"       ).transpose(2      ,      0      ,      1       )

						__SCREAMING_SNAKE_CASE						       =    [self._np_extract_fbank_features(UpperCAmelCase__       ) for waveform in input_features[0]]

						if isinstance(input_features[0]      ,      UpperCAmelCase__       ):
											__SCREAMING_SNAKE_CASE						       =    [np.asarray(UpperCAmelCase__      ,      dtype=np.floataa       ) for feature in input_features]
						else:
											__SCREAMING_SNAKE_CASE						       =    input_features

						if return_attention_mask:
											# rescale from sample (48000) to feature (3000)
											__SCREAMING_SNAKE_CASE						       =    padded_inputs["attention_mask"][:, :: self.hop_length]

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

						return padded_inputs




	def   UpperCAmelCase_				(			self      :    Optional[Any]       )	->     Dict:
						__SCREAMING_SNAKE_CASE						       =    copy.deepcopy(self.__dict__       )
						__SCREAMING_SNAKE_CASE						       =    self.__class__.__name__
						if "mel_filters" in output:
											del output["mel_filters"]
						return output




"""simple docstring"""







import re

import jax.numpy as jnp
from flax.traverse_util import flatten_dict, unflatten_dict
from jax.random import PRNGKey

from ..utils import logging


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




def        UpperCAmelCase__	(lowerCAmelCase_       ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    R"\w+[.]\d+"
					__SCREAMING_SNAKE_CASE						       =    re.findall(lowerCAmelCase_  ,						lowerCAmelCase_       )
					for pat in pats:
										__SCREAMING_SNAKE_CASE						       =    key.replace(lowerCAmelCase_  ,						"_".join(pat.split("."       )       )       )
					return key




def        UpperCAmelCase__	(lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_       ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    pt_tuple_key[:-1] + ("scale",)
					if (
					    any("norm" in str_ for str_ in pt_tuple_key       )
					    and (pt_tuple_key[-1] == "bias")
					    and (pt_tuple_key[:-1] + ("bias",) not in random_flax_state_dict)
					    and (pt_tuple_key[:-1] + ("scale",) in random_flax_state_dict)
					):
										__SCREAMING_SNAKE_CASE						       =    pt_tuple_key[:-1] + ("scale",)
										return renamed_pt_tuple_key, pt_tensor
					elif pt_tuple_key[-1] in ["weight", "gamma"] and pt_tuple_key[:-1] + ("scale",) in random_flax_state_dict:
										__SCREAMING_SNAKE_CASE						       =    pt_tuple_key[:-1] + ("scale",)
										return renamed_pt_tuple_key, pt_tensor

					# embedding
					if pt_tuple_key[-1] == "weight" and pt_tuple_key[:-1] + ("embedding",) in random_flax_state_dict:
										__SCREAMING_SNAKE_CASE						       =    pt_tuple_key[:-1] + ("embedding",)
										return renamed_pt_tuple_key, pt_tensor

					# conv layer
					__SCREAMING_SNAKE_CASE						       =    pt_tuple_key[:-1] + ("kernel",)
					if pt_tuple_key[-1] == "weight" and pt_tensor.ndim == 4:
										__SCREAMING_SNAKE_CASE						       =    pt_tensor.transpose(2  ,						3  ,						1  ,						0       )
										return renamed_pt_tuple_key, pt_tensor

					# linear layer
					__SCREAMING_SNAKE_CASE						       =    pt_tuple_key[:-1] + ("kernel",)
					if pt_tuple_key[-1] == "weight":
										__SCREAMING_SNAKE_CASE						       =    pt_tensor.T
										return renamed_pt_tuple_key, pt_tensor

					# old PyTorch layer norm weight
					__SCREAMING_SNAKE_CASE						       =    pt_tuple_key[:-1] + ("weight",)
					if pt_tuple_key[-1] == "gamma":
										return renamed_pt_tuple_key, pt_tensor

					# old PyTorch layer norm bias
					__SCREAMING_SNAKE_CASE						       =    pt_tuple_key[:-1] + ("bias",)
					if pt_tuple_key[-1] == "beta":
										return renamed_pt_tuple_key, pt_tensor

					return pt_tuple_key, pt_tensor




def        UpperCAmelCase__	(lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_=42       ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    {k: v.numpy() for k, v in pt_state_dict.items()}

					# Step 2: Since the model is stateless, get random Flax params
					__SCREAMING_SNAKE_CASE						       =    flax_model.init_weights(PRNGKey(lowerCAmelCase_       )       )

					__SCREAMING_SNAKE_CASE						       =    flatten_dict(lowerCAmelCase_       )
					__SCREAMING_SNAKE_CASE						       =    {}

					# Need to change some parameters name to match Flax names
					for pt_key, pt_tensor in pt_state_dict.items():
										__SCREAMING_SNAKE_CASE						       =    rename_key(lowerCAmelCase_       )
										__SCREAMING_SNAKE_CASE						       =    tuple(renamed_pt_key.split("."       )       )

										# Correctly rename weight parameters
										__SCREAMING_SNAKE_CASE       ,  __SCREAMING_SNAKE_CASE						       =    rename_key_and_reshape_tensor(lowerCAmelCase_  ,						lowerCAmelCase_  ,						lowerCAmelCase_       )

										if flax_key in random_flax_state_dict:
															if flax_tensor.shape != random_flax_state_dict[flax_key].shape:
																				raise ValueError(
																				    f"""PyTorch checkpoint seems to be incorrect. Weight {pt_key} was expected to be of shape """
																				    f"""{random_flax_state_dict[flax_key].shape}, but is {flax_tensor.shape}."""       )

        # also add unexpected weight so that warning is thrown
										__SCREAMING_SNAKE_CASE						       =    jnp.asarray(lowerCAmelCase_       )

					return unflatten_dict(lowerCAmelCase_       )
