
import importlib.metadata
import warnings
from copy import deepcopy

from packaging import version

from ..utils import logging
from .import_utils import is_accelerate_available, is_bitsandbytes_available


if is_bitsandbytes_available():
							import bitsandbytes as bnb
							import torch
							import torch.nn as nn

							from ..pytorch_utils import ConvaD

if is_accelerate_available():
							from accelerate import init_empty_weights
							from accelerate.utils import find_tied_parameters

UpperCamelCase_						:     int	       =							logging.get_logger(__name__)



def 		UpperCamelCase     (       _UpperCAmelCase  :   Optional[Any] ,			_UpperCAmelCase  :   Optional[Any] ,			_UpperCAmelCase  :   Optional[Any] ,			_UpperCAmelCase  :   Dict=None ,			_UpperCAmelCase  :   Dict=None  )   ->			Tuple:






						'''simple docstring'''

						if "." in tensor_name:
												_lowercase       :	Dict           =   tensor_name.split("."  )
												for split in splits[:-1]:
																		_lowercase       :	Optional[int]           =   getattr(_UpperCAmelCase ,			_UpperCAmelCase  )
																		if new_module is None:
																								raise ValueError(f"""{module} has no attribute {split}."""  )
																		_lowercase       :	List[Any]           =   new_module
												_lowercase       :	Dict           =   splits[-1]

						if tensor_name not in module._parameters and tensor_name not in module._buffers:
												raise ValueError(f"""{module} does not have a parameter or a buffer named {tensor_name}."""  )
						_lowercase       :	Dict           =   tensor_name in module._buffers
						_lowercase       :	Union[str, Any]           =   getattr(_UpperCAmelCase ,			_UpperCAmelCase  )

						if old_value.device == torch.device("meta"  ) and device not in ["meta", torch.device("meta"  )] and value is None:
												raise ValueError(f"""{tensor_name} is on the meta device, we need a `value` to put in on {device}."""  )

						_lowercase       :	int           =   False
						_lowercase       :	str           =   False
						if is_buffer or not is_bitsandbytes_available():
												_lowercase       :	int           =   False
												_lowercase       :	List[str]           =   False
						else:
												_lowercase       :	List[str]           =   hasattr(bnb.nn ,			"Params4bit"  ) and isinstance(module._parameters[tensor_name] ,			bnb.nn.Paramsabit  )
												_lowercase       :	int           =   isinstance(module._parameters[tensor_name] ,			bnb.nn.IntaParams  )

						if is_abit or is_abit:
												_lowercase       :	Optional[int]           =   module._parameters[tensor_name]
												if param.device.type != "cuda":
																		if value is None:
																								_lowercase       :	int           =   old_value.to(_UpperCAmelCase  )
																		elif isinstance(_UpperCAmelCase ,			torch.Tensor  ):
																								_lowercase       :	Dict           =   value.to("cpu"  )
																								if value.dtype == torch.inta:
																														_lowercase       :	str           =   version.parse(importlib.metadata.version("bitsandbytes"  )  ) > version.parse(
																														    "0.37.2"  )
																														if not is_abit_serializable:
																																				raise ValueError(
																																				    "Detected int8 weights but the version of bitsandbytes is not compatible with int8 serialization. "
																																				    "Make sure to download the latest `bitsandbytes` version. `pip install --upgrade bitsandbytes`."  )
																		else:
																								_lowercase       :	List[str]           =   torch.tensor(_UpperCAmelCase ,			device="cpu"  )

																		# Support models using `Conv1D` in place of `nn.Linear` (e.g. gpt2) by transposing the weight matrix prior to quantization.
																		# Since weights are saved in the correct "orientation", we skip transposing when loading.
																		if issubclass(module.source_cls ,			_UpperCAmelCase  ) and fpaa_statistics is None:
																								_lowercase       :	List[Any]           =   new_value.T

																		_lowercase       :	List[str]           =   old_value.__dict__
																		if is_abit:
																								_lowercase       :	Optional[Any]           =   bnb.nn.IntaParams(_UpperCAmelCase ,			requires_grad=_UpperCAmelCase ,			**_UpperCAmelCase  ).to(_UpperCAmelCase  )
																		elif is_abit:
																								_lowercase       :	Dict           =   bnb.nn.Paramsabit(_UpperCAmelCase ,			requires_grad=_UpperCAmelCase ,			**_UpperCAmelCase  ).to(_UpperCAmelCase  )

																		_lowercase       :	str           =   new_value
																		if fpaa_statistics is not None:
																								setattr(module.weight ,			"SCB" ,			fpaa_statistics.to(_UpperCAmelCase  )  )

						else:
												if value is None:
																		_lowercase       :	Optional[Any]           =   old_value.to(_UpperCAmelCase  )
												elif isinstance(_UpperCAmelCase ,			torch.Tensor  ):
																		_lowercase       :	Tuple           =   value.to(_UpperCAmelCase  )
												else:
																		_lowercase       :	Optional[Any]           =   torch.tensor(_UpperCAmelCase ,			device=_UpperCAmelCase  )

												if is_buffer:
																		_lowercase       :	str           =   new_value
												else:
																		_lowercase       :	int           =   nn.Parameter(_UpperCAmelCase ,			requires_grad=old_value.requires_grad  )
																		_lowercase       :	str           =   new_value



def 		UpperCamelCase     (       _UpperCAmelCase  :   Any ,			_UpperCAmelCase  :   str=None ,			_UpperCAmelCase  :   List[str]=None ,			_UpperCAmelCase  :   Tuple=None ,			_UpperCAmelCase  :   Tuple=False  )   ->			int:






						'''simple docstring'''

						for name, module in model.named_children():
												if current_key_name is None:
																		_lowercase       :	int           =   []
												current_key_name.append(_UpperCAmelCase  )

												if (isinstance(_UpperCAmelCase ,			nn.Linear  ) or isinstance(_UpperCAmelCase ,			_UpperCAmelCase  )) and name not in modules_to_not_convert:
																		# Check if the current key is not in the `modules_to_not_convert`
																		if not any(key in ".".join(_UpperCAmelCase  ) for key in modules_to_not_convert  ):
																								with init_empty_weights():
																														if isinstance(_UpperCAmelCase ,			_UpperCAmelCase  ):
																																				_lowercase      ,    _lowercase       :	Tuple           =   module.weight.shape
																														else:
																																				_lowercase       :	List[Any]           =   module.in_features
																																				_lowercase       :	List[str]           =   module.out_features

																														if quantization_config.quantization_method() == "llm_int8":
																																				_lowercase       :	Union[str, Any]           =   bnb.nn.LinearabitLt(
																																				    _UpperCAmelCase ,			_UpperCAmelCase ,			module.bias is not None ,			has_fpaa_weights=quantization_config.llm_inta_has_fpaa_weight ,			threshold=quantization_config.llm_inta_threshold ,			)
																																				_lowercase       :	Tuple           =   True
																														else:
																																				if (
																																				    quantization_config.llm_inta_skip_modules is not None
																																				    and name in quantization_config.llm_inta_skip_modules
																																				):
																																										pass
																																				else:
																																										_lowercase       :	List[Any]           =   bnb.nn.Linearabit(
																																										    _UpperCAmelCase ,			_UpperCAmelCase ,			module.bias is not None ,			quantization_config.bnb_abit_compute_dtype ,			compress_statistics=quantization_config.bnb_abit_use_double_quant ,			quant_type=quantization_config.bnb_abit_quant_type ,			)
																																										_lowercase       :	Optional[int]           =   True
                    # Store the module class in case we need to transpose the weight later
																														_lowercase       :	str           =   type(_UpperCAmelCase  )
																														# Force requires grad to False to avoid unexpected errors
																														model._modules[name].requires_grad_(_UpperCAmelCase  )
												if len(list(module.children()  )  ) > 0:
																		_lowercase      ,    _lowercase       :	List[str]           =   _replace_with_bnb_linear(
																		    _UpperCAmelCase ,			_UpperCAmelCase ,			_UpperCAmelCase ,			_UpperCAmelCase ,			has_been_replaced=_UpperCAmelCase ,			)
												# Remove the last key for recursion
												current_key_name.pop(-1  )
						return model, has_been_replaced



def 		UpperCamelCase     (       _UpperCAmelCase  :   Optional[int] ,			_UpperCAmelCase  :   Tuple=None ,			_UpperCAmelCase  :   Optional[int]=None ,			_UpperCAmelCase  :   int=None  )   ->			Optional[int]:






						'''simple docstring'''

						_lowercase       :	str           =   ["lm_head"] if modules_to_not_convert is None else modules_to_not_convert
						_lowercase      ,    _lowercase       :	Any           =   _replace_with_bnb_linear(
						    _UpperCAmelCase ,			_UpperCAmelCase ,			_UpperCAmelCase ,			_UpperCAmelCase  )

						if not has_been_replaced:
												logger.warning(
												    "You are loading your model in 8bit or 4bit but no linear modules were found in your model."
												    " Please double check your model architecture, or submit an issue on github if you think this is"
												    " a bug."  )

						return model



def 		UpperCamelCase     (       *_UpperCAmelCase  :   Optional[Any] ,			**_UpperCAmelCase  :   Dict  )   ->			Tuple:






						'''simple docstring'''

						warnings.warn(
						    "`replace_8bit_linear` will be deprecated in a future version, please use `replace_with_bnb_linear` instead" ,			_UpperCAmelCase ,			)
						return replace_with_bnb_linear(*_UpperCAmelCase ,			**_UpperCAmelCase  )



def 		UpperCamelCase     (       *_UpperCAmelCase  :   Dict ,			**_UpperCAmelCase  :   Dict  )   ->			int:






						'''simple docstring'''

						warnings.warn(
						    "`set_module_8bit_tensor_to_device` will be deprecated in a future version, please use `set_module_quantized_tensor_to_device` instead" ,			_UpperCAmelCase ,			)
						return set_module_quantized_tensor_to_device(*_UpperCAmelCase ,			**_UpperCAmelCase  )



def 		UpperCamelCase     (       _UpperCAmelCase  :   List[Any]  )   ->			Tuple:






						'''simple docstring'''

						_lowercase       :	Tuple           =   deepcopy(_UpperCAmelCase  )  # this has 0 cost since it is done inside `init_empty_weights` context manager`
						tied_model.tie_weights()

						_lowercase       :	List[Any]           =   find_tied_parameters(_UpperCAmelCase  )
						# For compatibility with Accelerate < 0.18
						if isinstance(_UpperCAmelCase ,			_UpperCAmelCase  ):
												_lowercase       :	Any           =   sum(list(tied_params.values()  ) ,			[]  ) + list(tied_params.keys()  )
						else:
												_lowercase       :	Tuple           =   sum(_UpperCAmelCase ,			[]  )
						_lowercase       :	List[Any]           =   len(_UpperCAmelCase  ) > 0

						# Check if it is a base model
						_lowercase       :	Any           =   not hasattr(_UpperCAmelCase ,			model.base_model_prefix  )

						# Ignore this for base models (BertModel, GPT2Model, etc.)
						if (not has_tied_params) and is_base_model:
												return []

						# otherwise they have an attached head
						_lowercase       :	int           =   list(model.named_children()  )
						_lowercase       :	str           =   [list_modules[-1][0]]

						# add last module together with tied weights
						_lowercase       :	Optional[int]           =   set(_UpperCAmelCase  ) - set(_UpperCAmelCase  )
						_lowercase       :	Optional[int]           =   list(set(_UpperCAmelCase  )  ) + list(_UpperCAmelCase  )

						# remove ".weight" from the keys
						_lowercase       :	str           =   [".weight", ".bias"]
						_lowercase       :	Optional[Any]           =   []
						for name in list_untouched:
												for name_to_remove in names_to_remove:
																		if name_to_remove in name:
																								_lowercase       :	Optional[Any]           =   name.replace(_UpperCAmelCase ,			""  )
												filtered_module_names.append(_UpperCAmelCase  )

						return filtered_module_names






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

import numpy as np

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


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




class   __lowercase						(						__snake_case       ):
	_A       =   ["input_values", "attention_mask"]






	def __init__(self				: int      ,	snake_case				: int = 1      ,	snake_case				: int = 1_6000      ,	snake_case				: float = 0.0      ,	snake_case				: bool = False      ,	snake_case				: int = 80      ,	snake_case				: int = 16      ,	snake_case				: int = 64      ,	snake_case				: str = "hann_window"      ,	snake_case				: float = 1.0      ,	snake_case				: float = 80      ,	snake_case				: float = 7600      ,	snake_case				: float = 1e-10      ,	snake_case				: int = 2      ,	snake_case				: bool = True      ,	**snake_case				: Any      ,	) ->			str:
							super().__init__(feature_size=snake_case      ,	sampling_rate=snake_case      ,	padding_value=snake_case      ,	**snake_case    )
							_lowercase       :	int           =   do_normalize
							_lowercase       :	Any           =   return_attention_mask

							_lowercase       :	str           =   num_mel_bins
							_lowercase       :	str           =   hop_length
							_lowercase       :	List[Any]           =   win_length
							_lowercase       :	Tuple           =   win_function
							_lowercase       :	Any           =   frame_signal_scale
							_lowercase       :	Any           =   fmin
							_lowercase       :	Any           =   fmax
							_lowercase       :	List[str]           =   mel_floor
							_lowercase       :	str           =   reduction_factor

							_lowercase       :	str           =   win_length * sampling_rate // 1000
							_lowercase       :	Optional[int]           =   hop_length * sampling_rate // 1000
							_lowercase       :	Tuple           =   optimal_fft_length(self.sample_size    )
							_lowercase       :	int           =   (self.n_fft // 2) + 1

							_lowercase       :	Tuple           =   window_function(window_length=self.sample_size      ,	name=self.win_function      ,	periodic=snake_case    )

							_lowercase       :	Optional[int]           =   mel_filter_bank(
							    num_frequency_bins=self.n_freqs      ,	num_mel_filters=self.num_mel_bins      ,	min_frequency=self.fmin      ,	max_frequency=self.fmax      ,	sampling_rate=self.sampling_rate      ,	norm="slaney"      ,	mel_scale="slaney"      ,	)

							if frame_signal_scale != 1.0:
													warnings.warn(
													    "The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"      ,	snake_case      ,	)
							if reduction_factor != 2.0:
													warnings.warn(
													    "The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"      ,	snake_case      ,	)






	@staticmethod
	# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
	def       _a(snake_case				: List[np.ndarray]      ,	snake_case				: List[np.ndarray]      ,	snake_case				: float = 0.0    ) ->			List[np.ndarray]:
							if attention_mask is not None:
													_lowercase       :	List[str]           =   np.array(snake_case      ,	np.intaa    )
													_lowercase       :	Dict           =   []

													for vector, length in zip(snake_case      ,	attention_mask.sum(-1    )    ):
																			_lowercase       :	Tuple           =   (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7    )
																			if length < normed_slice.shape[0]:
																									_lowercase       :	List[str]           =   padding_value

																			normed_input_values.append(snake_case    )
							else:
													_lowercase       :	Dict           =   [(x - x.mean()) / np.sqrt(x.var() + 1e-7    ) for x in input_values]

							return normed_input_values






	def       _a(self				: Optional[Any]      ,	snake_case				: np.ndarray      ,	) ->			np.ndarray:
							_lowercase       :	Tuple           =   spectrogram(
							    snake_case      ,	window=self.window      ,	frame_length=self.sample_size      ,	hop_length=self.sample_stride      ,	fft_length=self.n_fft      ,	mel_filters=self.mel_filters      ,	mel_floor=self.mel_floor      ,	log_mel="log10"      ,	)
							return log_mel_spec.T






	def __call__(self				: Tuple      ,	snake_case				: Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None      ,	snake_case				: Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None      ,	snake_case				: Union[bool, str, PaddingStrategy] = False      ,	snake_case				: Optional[int] = None      ,	snake_case				: bool = False      ,	snake_case				: Optional[int] = None      ,	snake_case				: Optional[bool] = None      ,	snake_case				: Optional[Union[str, TensorType]] = None      ,	snake_case				: Optional[int] = None      ,	**snake_case				: Union[str, Any]      ,	) ->			BatchFeature:
							if audio is None and audio_target is None:
													raise ValueError("You must provide either `audio` or `audio_target` values."    )

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

							if audio is not None:
													_lowercase       :	Union[str, Any]           =   self._process_audio(
													    snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	**snake_case      ,	)
							else:
													_lowercase       :	str           =   None

							if audio_target is not None:
													_lowercase       :	Union[str, Any]           =   self._process_audio(
													    snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	snake_case      ,	**snake_case      ,	)

													if inputs is None:
																			return inputs_target
													else:
																			_lowercase       :	int           =   inputs_target["input_values"]
																			_lowercase       :	Optional[int]           =   inputs_target.get("attention_mask"    )
																			if decoder_attention_mask is not None:
																									_lowercase       :	List[str]           =   decoder_attention_mask

							return inputs






	def       _a(self				: Optional[Any]      ,	snake_case				: Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]      ,	snake_case				: bool = False      ,	snake_case				: Union[bool, str, PaddingStrategy] = False      ,	snake_case				: Optional[int] = None      ,	snake_case				: bool = False      ,	snake_case				: Optional[int] = None      ,	snake_case				: Optional[bool] = None      ,	snake_case				: Optional[Union[str, TensorType]] = None      ,	**snake_case				: int      ,	) ->			BatchFeature:
							_lowercase       :	int           =   isinstance(snake_case      ,	np.ndarray    ) and len(speech.shape    ) > 1
							if is_batched_numpy and len(speech.shape    ) > 2:
													raise ValueError(F"""Only mono-channel audio is supported for input to {self}"""    )
							_lowercase       :	List[str]           =   is_batched_numpy or (
							    isinstance(snake_case      ,	(list, tuple)    ) and (isinstance(speech[0]      ,	(np.ndarray, tuple, list)    ))
							)

							if is_batched:
													_lowercase       :	List[Any]           =   [np.asarray(snake_case      ,	dtype=np.floataa    ) for speech in speech]
							elif not is_batched and not isinstance(snake_case      ,	np.ndarray    ):
													_lowercase       :	Optional[Any]           =   np.asarray(snake_case      ,	dtype=np.floataa    )
							elif isinstance(snake_case      ,	np.ndarray    ) and speech.dtype is np.dtype(np.floataa    ):
													_lowercase       :	Optional[int]           =   speech.astype(np.floataa    )

							# always return batch
							if not is_batched:
													_lowercase       :	int           =   [speech]

							# needed to make pad() work on spectrogram inputs
							_lowercase       :	Tuple           =   self.feature_size

							# convert into correct format for padding
							if is_target:
													_lowercase       :	List[str]           =   [self._extract_mel_features(snake_case    ) for waveform in speech]
													_lowercase       :	int           =   BatchFeature({"input_values": features}    )
													_lowercase       :	str           =   self.num_mel_bins
							else:
													_lowercase       :	Optional[int]           =   BatchFeature({"input_values": speech}    )

							_lowercase       :	List[Any]           =   self.pad(
							    snake_case      ,	padding=snake_case      ,	max_length=snake_case      ,	truncation=snake_case      ,	pad_to_multiple_of=snake_case      ,	return_attention_mask=snake_case      ,	**snake_case      ,	)

							_lowercase       :	Optional[Any]           =   feature_size_hack

							# convert input values to correct format
							_lowercase       :	int           =   padded_inputs["input_values"]
							if not isinstance(input_values[0]      ,	np.ndarray    ):
													_lowercase       :	Any           =   [np.asarray(snake_case      ,	dtype=np.floataa    ) for array in input_values]
							elif (
							    not isinstance(snake_case      ,	np.ndarray    )
							    and isinstance(input_values[0]      ,	np.ndarray    )
							    and input_values[0].dtype is np.dtype(np.floataa    )
							):
													_lowercase       :	Union[str, Any]           =   [array.astype(np.floataa    ) for array in input_values]
							elif isinstance(snake_case      ,	np.ndarray    ) and input_values.dtype is np.dtype(np.floataa    ):
													_lowercase       :	Tuple           =   input_values.astype(np.floataa    )

							# convert attention_mask to correct format
							_lowercase       :	Union[str, Any]           =   padded_inputs.get("attention_mask"    )
							if attention_mask is not None:
													_lowercase       :	str           =   [np.asarray(snake_case      ,	dtype=np.intaa    ) for array in attention_mask]

							# zero-mean and unit-variance normalization
							if not is_target and self.do_normalize:
													_lowercase       :	Dict           =   (
													    attention_mask
													    if self._get_padding_strategies(snake_case      ,	max_length=snake_case    ) is not PaddingStrategy.DO_NOT_PAD
													    else None
													)
													_lowercase       :	List[Any]           =   self.zero_mean_unit_var_norm(
													    padded_inputs["input_values"]      ,	attention_mask=snake_case      ,	padding_value=self.padding_value    )

							if return_tensors is not None:
													_lowercase       :	Any           =   padded_inputs.convert_to_tensors(snake_case    )

							return padded_inputs






	def       _a(self				: Dict    ) ->			Dict[str, Any]:
							_lowercase       :	str           =   super().to_dict()

							# Don't serialize these as they are derived from the other properties.
							_lowercase       :	Optional[Any]           =   ["window", "mel_filters", "sample_size", "sample_stride", "n_fft", "n_freqs"]
							for name in names:
													if name in output:
																			del output[name]

							return output




