




"""simple docstring"""





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

import numpy as np
import torch

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				=							logging.get_logger(__name__)



class   a__					(						__magic_name__ ):
			lowercase_		=						["input_features", "is_longer"]


			def __init__(   self   :     List[str]   ,	UpperCamelCase_   :     Dict=64   ,	UpperCamelCase_   :     Tuple=48000   ,	UpperCamelCase_   :     List[Any]=480   ,	UpperCamelCase_   :     List[str]=10   ,	UpperCamelCase_   :     str=1024   ,	UpperCamelCase_   :     List[str]=0.0   ,	UpperCamelCase_   :     Optional[int]=False   ,	UpperCamelCase_   :     float = 0   ,	UpperCamelCase_   :     float = 14000   ,	UpperCamelCase_   :     int = None   ,	UpperCamelCase_   :     str = "fusion"   ,	UpperCamelCase_   :     str = "repeatpad"   ,	**UpperCamelCase_   :     Optional[Any]   ,	):


						"""simple docstring"""




						super().__init__(
						    feature_size=UpperCamelCase_   ,	sampling_rate=UpperCamelCase_   ,	padding_value=UpperCamelCase_   ,	return_attention_mask=UpperCamelCase_   ,	**UpperCamelCase_   ,	)
						__UpperCAmelCase						:			Union[str, Any]												=       top_db
						__UpperCAmelCase						:			Optional[Any]												=       truncation
						__UpperCAmelCase						:			str												=       padding
						__UpperCAmelCase						:			int												=       fft_window_size
						__UpperCAmelCase						:			str												=       (fft_window_size >> 1) + 1
						__UpperCAmelCase						:			List[Any]												=       hop_length
						__UpperCAmelCase						:			Optional[Any]												=       max_length_s
						__UpperCAmelCase						:			Tuple												=       max_length_s * sampling_rate
						__UpperCAmelCase						:			str												=       sampling_rate
						__UpperCAmelCase						:			int												=       frequency_min
						__UpperCAmelCase						:			Optional[Any]												=       frequency_max
						__UpperCAmelCase						:			Any												=       mel_filter_bank(
						    num_frequency_bins=self.nb_frequency_bins   ,	num_mel_filters=UpperCamelCase_   ,	min_frequency=UpperCamelCase_   ,	max_frequency=UpperCamelCase_   ,	sampling_rate=UpperCamelCase_   ,	norm=UpperCamelCase_   ,	mel_scale="htk"   ,	)
						__UpperCAmelCase						:			Any												=       mel_filter_bank(
						    num_frequency_bins=self.nb_frequency_bins   ,	num_mel_filters=UpperCamelCase_   ,	min_frequency=UpperCamelCase_   ,	max_frequency=UpperCamelCase_   ,	sampling_rate=UpperCamelCase_   ,	norm="slaney"   ,	mel_scale="slaney"   ,	)


			def    a_  (   self   :     Dict):


						"""simple docstring"""




						__UpperCAmelCase						:			Dict												=       copy.deepcopy(self.__dict__)
						__UpperCAmelCase						:			str												=       self.__class__.__name__
						if "mel_filters" in output:
									del output["mel_filters"]
						if "mel_filters_slaney" in output:
									del output["mel_filters_slaney"]
						return output


			def    a_  (   self   :     int   ,	UpperCamelCase_   :     np.array   ,	UpperCamelCase_   :     Optional[np.array] = None):


						"""simple docstring"""




						__UpperCAmelCase						:			List[Any]												=       spectrogram(
						    UpperCamelCase_   ,	window_function(self.fft_window_size   ,	"hann")   ,	frame_length=self.fft_window_size   ,	hop_length=self.hop_length   ,	power=2.0   ,	mel_filters=UpperCamelCase_   ,	log_mel="dB"   ,	)
						return log_mel_spectrogram.T


			def    a_  (   self   :     List[Any]   ,	UpperCamelCase_   :     List[Any]   ,	UpperCamelCase_   :     Optional[int]   ,	UpperCamelCase_   :     int):


						"""simple docstring"""




						__UpperCAmelCase						:			Optional[Any]												=       np.array_split(list(range(0   ,	total_frames - chunk_frames + 1))   ,	3)
						if len(ranges[1]) == 0:
									# if the audio is too short, we just use the first chunk
									__UpperCAmelCase						:			str												=       [0]
						if len(ranges[2]) == 0:
									# if the audio is too short, we just use the first chunk
									__UpperCAmelCase						:			Dict												=       [0]
						# randomly choose index for each part
						__UpperCAmelCase						:			Dict												=       np.random.choice(ranges[0])
						__UpperCAmelCase						:			List[str]												=       np.random.choice(ranges[1])
						__UpperCAmelCase						:			List[Any]												=       np.random.choice(ranges[2])

						__UpperCAmelCase						:			List[Any]												=       mel[idx_front : idx_front + chunk_frames, :]
						__UpperCAmelCase						:			List[str]												=       mel[idx_middle : idx_middle + chunk_frames, :]
						__UpperCAmelCase						:			List[str]												=       mel[idx_back : idx_back + chunk_frames, :]

						__UpperCAmelCase						:			Tuple												=       torch.tensor(mel[None, None, :])
						__UpperCAmelCase						:			Union[str, Any]												=       torch.nn.functional.interpolate(
						    UpperCamelCase_   ,	size=[chunk_frames, 64]   ,	mode="bilinear"   ,	align_corners=UpperCamelCase_)
						__UpperCAmelCase						:			Union[str, Any]												=       mel_shrink[0][0].numpy()
						__UpperCAmelCase						:			Optional[int]												=       np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]   ,	axis=0)
						return mel_fusion


			def    a_  (   self   :     Optional[Any]   ,	UpperCamelCase_   :     np.array   ,	UpperCamelCase_   :     Any   ,	UpperCamelCase_   :     Tuple   ,	UpperCamelCase_   :     Optional[Any]):


						"""simple docstring"""




						if waveform.shape[0] > max_length:
									if truncation == "rand_trunc":
												__UpperCAmelCase						:			List[str]												=       True
												# random crop to max_length (for compatibility) -> this should be handled by self.pad
												__UpperCAmelCase						:			List[Any]												=       len(UpperCamelCase_) - max_length
												__UpperCAmelCase						:			int												=       np.random.randint(0   ,	overflow + 1)
												__UpperCAmelCase						:			Union[str, Any]												=       waveform[idx : idx + max_length]
												__UpperCAmelCase						:			Union[str, Any]												=       self._np_extract_fbank_features(UpperCamelCase_   ,	self.mel_filters_slaney)[None, :]
									elif truncation == "fusion":
												__UpperCAmelCase						:			Any												=       self._np_extract_fbank_features(UpperCamelCase_   ,	self.mel_filters)
												__UpperCAmelCase						:			Dict												=       max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
												__UpperCAmelCase						:			Tuple												=       mel.shape[0]
												if chunk_frames == total_frames:
															# there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
															# In this case, we just use the whole audio.
															__UpperCAmelCase						:			List[str]												=       np.stack([mel, mel, mel, mel]   ,	axis=0)
															__UpperCAmelCase						:			Any												=       False
												else:
															__UpperCAmelCase						:			List[str]												=       self._random_mel_fusion(UpperCamelCase_   ,	UpperCamelCase_   ,	UpperCamelCase_)
															__UpperCAmelCase						:			Union[str, Any]												=       True
									else:
												raise NotImplementedError(F"data_truncating {truncation} not implemented")

						else:
									__UpperCAmelCase						:			Optional[Any]												=       False
									# only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
									if waveform.shape[0] < max_length:
												if padding == "repeat":
															__UpperCAmelCase						:			Tuple												=       int(max_length / len(UpperCamelCase_))
															__UpperCAmelCase						:			List[str]												=       np.stack(np.tile(UpperCamelCase_   ,	n_repeat + 1))[:max_length]
												if padding == "repeatpad":
															__UpperCAmelCase						:			Union[str, Any]												=       int(max_length / len(UpperCamelCase_))
															__UpperCAmelCase						:			Optional[Any]												=       np.stack(np.tile(UpperCamelCase_   ,	UpperCamelCase_))
												__UpperCAmelCase						:			int												=       np.pad(UpperCamelCase_   ,	(0, max_length - waveform.shape[0])   ,	mode="constant"   ,	constant_values=0)

									if truncation == "fusion":
												__UpperCAmelCase						:			Any												=       self._np_extract_fbank_features(UpperCamelCase_   ,	self.mel_filters)
												__UpperCAmelCase						:			List[Any]												=       np.stack([input_mel, input_mel, input_mel, input_mel]   ,	axis=0)
									else:
												__UpperCAmelCase						:			Optional[int]												=       self._np_extract_fbank_features(UpperCamelCase_   ,	self.mel_filters_slaney)[None, :]

						return input_mel, longer


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


						"""simple docstring"""




						__UpperCAmelCase						:			int												=       truncation if truncation is not None else self.truncation
						__UpperCAmelCase						:			Optional[Any]												=       padding if padding else self.padding

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

						__UpperCAmelCase						:			List[str]												=       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}")
						__UpperCAmelCase						:			str												=       is_batched_numpy or (
						    isinstance(UpperCamelCase_   ,	(list, tuple)) and (isinstance(raw_speech[0]   ,	(np.ndarray, tuple, list)))
						)

						if is_batched:
									__UpperCAmelCase						:			Dict												=       [np.asarray(UpperCamelCase_   ,	dtype=np.floataa) for speech in raw_speech]
						elif not is_batched and not isinstance(UpperCamelCase_   ,	np.ndarray):
									__UpperCAmelCase						:			Tuple												=       np.asarray(UpperCamelCase_   ,	dtype=np.floataa)
						elif isinstance(UpperCamelCase_   ,	np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
									__UpperCAmelCase						:			Optional[int]												=       raw_speech.astype(np.floataa)

						# always return batch
						if not is_batched:
									__UpperCAmelCase						:			int												=       [np.asarray(UpperCamelCase_)]

						# convert to mel spectrogram, truncate and pad if needed.
						__UpperCAmelCase						:			Optional[int]												=       [
						    self._get_input_mel(UpperCamelCase_   ,	max_length if max_length else self.nb_max_samples   ,	UpperCamelCase_   ,	UpperCamelCase_)
						    for waveform in raw_speech
						]

						__UpperCAmelCase						:			Tuple												=       []
						__UpperCAmelCase						:			List[Any]												=       []
						for mel, longer in padded_inputs:
									input_mel.append(UpperCamelCase_)
									is_longer.append(UpperCamelCase_)

						if truncation == "fusion" and sum(UpperCamelCase_) == 0:
									# if no audio is longer than 10s, then randomly select one audio to be longer
									__UpperCAmelCase						:			Any												=       np.random.randint(0   ,	len(UpperCamelCase_))
									__UpperCAmelCase						:			Optional[int]												=       True

						if isinstance(input_mel[0]   ,	UpperCamelCase_):
									__UpperCAmelCase						:			Tuple												=       [np.asarray(UpperCamelCase_   ,	dtype=np.floataa) for feature in input_mel]

						# is_longer is a list of bool
						__UpperCAmelCase						:			List[str]												=       [[longer] for longer in is_longer]

						__UpperCAmelCase						:			Optional[int]												=       {"input_features": input_mel, "is_longer": is_longer}
						__UpperCAmelCase						:			Optional[int]												=       BatchFeature(UpperCamelCase_)

						if return_tensors is not None:
									__UpperCAmelCase						:			Any												=       input_features.convert_to_tensors(UpperCamelCase_)

						return input_features






# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import copy
import importlib.metadata
import json
import os
from dataclasses import dataclass
from typing import Any, Dict, Union

from packaging import version

from ..utils import is_torch_available, logging


if is_torch_available():
      import torch


snake_case__				:	int					       =					logging.get_logger(__name__)





@dataclass
class   SCREAMING_SNAKE_CASE_			:




       '''simple docstring'''



       def __init__(						self				:				int		,				__a				:				Optional[Any]=False		,				__a				:				Tuple=False		,				__a				:				Tuple=6.0		,				__a				:				str=None		,				__a				:				List[str]=False		,				__a				:				str=False		,				__a				:				Optional[Any]=None		,				__a				:				Dict="fp4"		,				__a				:				Optional[int]=False		,				**__a				:				Dict		,				)					->str:
          lowerCamelCase_						:   str		= load_in_abit
          lowerCamelCase_						:   Union[str, Any]		= load_in_abit
          lowerCamelCase_						:   Tuple		= llm_inta_threshold
          lowerCamelCase_						:   int		= llm_inta_skip_modules
          lowerCamelCase_						:   List[str]		= llm_inta_enable_fpaa_cpu_offload
          lowerCamelCase_						:   Optional[Any]		= llm_inta_has_fpaa_weight
          lowerCamelCase_						:   Dict		= bnb_abit_quant_type
          lowerCamelCase_						:   int		= bnb_abit_use_double_quant

          if bnb_abit_compute_dtype is None:
             lowerCamelCase_						:   int		= torch.floataa
          elif isinstance(__a		,				__a							):
             lowerCamelCase_						:   Tuple		= getattr(__a		,				__a							)
          elif isinstance(__a		,				torch.dtype							):
             lowerCamelCase_						:   List[str]		= bnb_abit_compute_dtype
          else:
             raise ValueError("""bnb_4bit_compute_dtype must be a string or a torch.dtype"""							)

          self.post_init()



       def 				_lowerCAmelCase							(						self				:				Any							)					->Union[str, Any]:
          if not isinstance(self.llm_inta_threshold		,				__a							):
             raise ValueError("""llm_int8_threshold must be a float"""							)

          if self.llm_inta_skip_modules is not None and not isinstance(self.llm_inta_skip_modules		,				__a							):
             raise ValueError("""llm_int8_skip_modules must be a list of strings"""							)
          if not isinstance(self.llm_inta_enable_fpaa_cpu_offload		,				__a							):
             raise ValueError("""llm_int8_enable_fp32_cpu_offload must be a boolean"""							)

          if not isinstance(self.llm_inta_has_fpaa_weight		,				__a							):
             raise ValueError("""llm_int8_has_fp16_weight must be a boolean"""							)

          if self.bnb_abit_compute_dtype is not None and not isinstance(self.bnb_abit_compute_dtype		,				torch.dtype							):
             raise ValueError("""bnb_4bit_compute_dtype must be torch.dtype"""							)

          if not isinstance(self.bnb_abit_quant_type		,				__a							):
             raise ValueError("""bnb_4bit_quant_type must be a string"""							)

          if not isinstance(self.bnb_abit_use_double_quant		,				__a							):
             raise ValueError("""bnb_4bit_use_double_quant must be a boolean"""							)

          if self.load_in_abit and not version.parse(importlib.metadata.version("""bitsandbytes"""							)							) >= version.parse(
              """0.39.0"""							):
             raise ValueError(
                 """4 bit quantization requires bitsandbytes>=0.39.0 - please upgrade your bitsandbytes version"""							)



       def 				_lowerCAmelCase							(						self				:				Optional[int]							)					->int:
          return self.load_in_abit or self.load_in_abit



       def 				_lowerCAmelCase							(						self				:				List[str]							)					->List[Any]:
          if self.load_in_abit:
             return "llm_int8"
          elif self.load_in_abit and self.bnb_abit_quant_type == "fp4":
             return "fp4"
          elif self.load_in_abit and self.bnb_abit_quant_type == "nf4":
             return "nf4"
          else:
             return None



       @classmethod
       def 				_lowerCAmelCase							(						cls				:				int		,				__a				:				Dict		,				__a				:				str		,				**__a				:				Any							)					->Optional[Any]:
          lowerCamelCase_						:   Optional[Any]		= cls(**__a							)

          lowerCamelCase_						:   List[Any]		= []
          for key, value in kwargs.items():
             if hasattr(__a		,				__a							):
                setattr(__a		,				__a		,				__a							)
                to_remove.append(__a							)
          for key in to_remove:
             kwargs.pop(__a		,				__a							)

          if return_unused_kwargs:
             return config, kwargs
          else:
             return config



       def 				_lowerCAmelCase							(						self				:				Optional[Any]		,				__a				:				Union[str, os.PathLike]							)					->Any:
          with open(__a		,				"""w"""		,				encoding="""utf-8"""							) as writer:
             lowerCamelCase_						:   List[Any]		= self.to_dict()
             lowerCamelCase_						:   Dict		= json.dumps(__a		,				indent=2		,				sort_keys=__a							) + """\n"""

             writer.write(__a							)



       def 				_lowerCAmelCase							(						self				:				Tuple							)					->Dict[str, Any]:
          lowerCamelCase_						:   Dict		= copy.deepcopy(self.__dict__							)
          lowerCamelCase_						:   Tuple		= str(output["""bnb_4bit_compute_dtype"""]							).split("""."""							)[1]

          return output



       def __repr__(						self				:				List[str]							)					->Any:
          return F'''{self.__class__.__name__} {self.to_json_string()}'''



       def 				_lowerCAmelCase							(						self				:				Optional[int]		,				__a				:				bool = True							)					->str:
          if use_diff is True:
             lowerCamelCase_						:   Dict		= self.to_diff_dict()
          else:
             lowerCamelCase_						:   Any		= self.to_dict()
          return json.dumps(__a		,				indent=2		,				sort_keys=__a							) + "\n"




       def 				_lowerCAmelCase							(						self				:				Optional[Any]							)					->Dict[str, Any]:
          lowerCamelCase_						:   Any		= self.to_dict()

          # get the default config dict
          lowerCamelCase_						:   List[str]		= BitsAndBytesConfig().to_dict()

          lowerCamelCase_						:   str		= {}

          # only serialize values that differ from the default config
          for key, value in config_dict.items():
             if value != default_config_dict[key]:
                lowerCamelCase_						:   List[Any]		= value

          return serializable_config_dict





